[Bug]: Pool is created but not stored, preventing reconnection on connection drop

[levish0]

What happened?

AmqpBackend creates a deadpool_lapin::Pool during initialization but discards it immediately, storing only a single Channel. This defeats the purpose of using a connection pool and makes recovery from connection drops impossible.

When a connection drops:

1. The stored Channel becomes invalid
2. heartbeat() detects disconnection and returns an error
3. No recovery is possible because the Pool was discarded
4. Worker stops permanently

AmqpBackend struct (src/lib.rs:61-68)

pub struct AmqpBackend<M, Codec> {
    channel: Channel,       // Only stores a single channel
    queue: lapin::Queue,
    message_type: PhantomData<M>,
    config: Config,
    sink: sink::AmqpSink<M, Codec>,
    // Pool is NOT stored!
}

new_from_addr_with_config (src/lib.rs:258-287)

pub async fn new_from_addr_with_config<S: AsRef<str>>(
    addr: S,
    config: Config,
) -> Result<AmqpBackend<M, JsonCodec<Vec<u8>>>, lapin::Error> {
    let manager = Manager::new(addr.as_ref(), ConnectionProperties::default());
    let pool: Pool = deadpool::managed::Pool::builder(manager)
        .max_size(10)
        .build()
        .map_err(|error| { /* ... */ })?;

    let amqp_conn = pool.get().await.map_err(|error| { /* ... */ })?;
    let channel = amqp_conn.create_channel().await?;
    let queue = channel.queue_declare(/* ... */).await?;

    Ok(Self::new_with_config(channel, queue, config))
    // ^^^ Pool is dropped here. Only channel is passed.
}

heartbeat (src/lib.rs:94-118)

fn heartbeat(&self, worker: &WorkerContext) -> Self::Beat {
    // ...
    let is_connected = channel.status().connected();
    if !is_connected {
        // Only returns error, no reconnection attempt
        Some((Err(ErrorKind::IOError(/* ... */)), /* ... */))
    }
    // ...
}

Impact

• Any network hiccup kills the worker permanently
• deadpool_lapin provides connection pooling and recycling, but none of it is used
• Long-running workers will eventually fail and never recover

Expected behavior

When a connection drops:

1. Backend should detect the disconnection
2. Get a new connection from the pool
3. Create a fresh channel
4. Continue processing

Steps to reproduce

1. Start a worker using AmqpBackend::new_from_addr()
2. Restart RabbitMQ (or any network interruption)
3. Connection is restored

Expected: Worker recovers and continues processing
Actual: Worker stays dead, never reconnects

Version

0.7.x

Environment

• OS: Windows 11
• Rust version: 1.92.0
• Cargo version: 1.92.0

[levish0]

I'd like to propose a fix for this. Here's my approach:

Root Cause

The pool is created but immediately discarded:

// Before (new_from_addr_with_config)
let pool: Pool = Pool::builder(manager).max_size(10).build()?;
let amqp_conn = pool.get().await?;
let channel = amqp_conn.create_channel().await?;
Ok(Self::new_with_config(channel, queue, config))
// ^^^ pool is dropped here, only channel is kept

Solution

1. Store Pool in struct

// Before
pub struct AmqpBackend<M, Codec> {
    channel: Channel,
    queue: lapin::Queue,
    // ...
}

// After
pub struct AmqpBackend<M, Codec> {
    pool: Pool,
    channel: Arc<RwLock<Channel>>,
    queue_name: String,  // Queue object not needed, only name
    // ...
}

2. Add reconnection logic in heartbeat

fn heartbeat(&self, worker: &WorkerContext) -> Self::Beat {
    // ...
    let is_connected = {
        let ch = channel.read().await;
        ch.status().connected()
    };
    if !is_connected {
        // Try to reconnect using pool
        match Self::reconnect_channel(&pool, &channel).await {
            Ok(()) => {
                tracing::info!("Reconnected channel for worker {}", worker.name());
            }
            Err(e) => {
                // Don't fail - will retry on next heartbeat
                tracing::warn!("Failed to reconnect: {:?}. Will retry.", e);
            }
        }
    }
    // Always return Ok to keep worker alive
    Some((Ok(()), (pool, channel, worker, config)))
}

Reconnection failures don't kill the worker. It keeps retrying every heartbeat interval (default 60s) until the broker is available again.

3. Reconnect helper

async fn reconnect_channel(
    pool: &Pool,
    channel: &Arc<RwLock<Channel>>,
) -> Result<(), Error> {
    let mut ch = channel.write().await;
    // Double-check: another task might have reconnected already
    if ch.status().connected() {
        return Ok(());
    }
    let conn = pool.get().await?;  // deadpool handles recycling
    *ch = conn.create_channel().await?;
    Ok(())
}

4. Update sink/ack to use RwLock

// Before
channel.basic_publish(...).await?;

// After
let ch = channel.read().await;
ch.basic_publish(...).await?;

New Dependency

[dependencies]
tokio = { version = "1", features = ["sync"] }

API Changes

Before	After
new(channel, queue)	new_with_pool(pool, channel, queue_name)
new_with_config(...)	new_with_pool_and_config(...)
new_from_addr(addr)	unchanged (now stores pool internally)
queue() -> &Queue	queue_name() -> &str
channel() -> &Channel	removed (channel is behind RwLock)

misc

• When channel dies mid-processing, inflight messages cannot be ACK'd (AMQP protocol limitation). The broker will requeue them after timeout.
• Arc<RwLock<Channel>> is needed because channel must be shared across heartbeat/poll/sink/ack AND replaceable on reconnect.

[geofmureithi]

I am open to this change. Here are some general thoughts:

1. Heartbeat should fail if re-connection fails
2. You might want to use futures instead of tokio

But basically go for it and then we can discuss specifics.

[levish0]

Looking at apalis-redis, it appears that errors are wrapped and the heartbeat stream continues rather than failing. This allows the worker to automatically recover when the connection is restored.

If the heartbeat doesn't terminate the connection entirely, the worker can continue processing jobs seamlessly when the broker comes back up, without requiring any manual intervention or restart.

---

I will look into using futures instead of tokio and update accordingly.

[levish0]

I noticed that apalis-redis relies on Redis's ConnectionManager which handles automatic reconnection transparently. Looking at the implementation

It uses macros like reconnect_if_dropped! and reconnect_if_io_error! to detect connection issues after every command. When a command fails due to connection issues, it triggers a background reconnection -> the first command fails, but subsequent commands will work once reconnection completes. Since AMQP doesn't have an equivalent to ConnectionManager.