PRECONDITION_FAILED error occurs in long-running consumers

[ohno104]

Hi! First of all, thank you for your library!

I'm new to Rust and currently using lapin = 2.5.0.
I'm trying to run multiple long-lived consumers that can automatically reconnect and gracefully shut down when needed. However, after running for an extended period, I occasionally encounter a PRECONDITION_FAILED error.

"error": "AMQPError { kind: Soft(PRECONDITIONFAILED), message: ShortString(\"PRECONDITION_FAILED - delivery acknowledgement on channel 1 timed out. Timeout value used: 1800000 ms. This timeout value can be configured, see consumers doc guide to learn more\") }",
"method": "Close { reply_code: 406, reply_text: ShortString(\"PRECONDITION_FAILED - delivery acknowledgement on channel 1 timed out. Timeout value used: 1800000 ms. This timeout value can be configured, see consumers doc guide to learn more\"), class_id: 0, method_id: 0 }"

Here is a simplified version of my code:

#[tokio::main]
async fn main() {
    let worker = Arc::new(Worker::new(pool)); // pool: bb8::Pool::builder()
    let mut join_handles = vec![];
    let cancel_token = CancellationToken::new();

    for i in 1..=5 {
        let join_handle = tokio::spawn(async move {
            if let Err(e) = worker.start(&cancel_token).await {
                error!(e);
            }
        });
        join_handles.push(join_handle);
    }

    signal::ctrl_c().await.expect("ctrl_c error");
    cancel_token.cancel();
    
    tokio::select! {
        _ = sleep(Duration::from_secs(5)) => {
                for handle in join_handles.iter_mut() {
                    handle.abort()
                }
                info!("forcefully shutdown");
            }

        _ = join_all(join_handles.iter_mut()) => {
            info!("tasks completed");
        }
    }
}

impl Worker {
    pub async fn start(&self, cancel_token: &CancellationToken) {
        let (mut channel, mut consumer) = self.get_connect().await;

        loop {
            if !channel.status().connected() || !consumer.state().is_active() {
                (channel, consumer) = self.get_connect().await;
            }

            tokio::select! {
                _ = cancel_token.cancelled() => {
                    if channel.status().connected(){
                        channel.basic_cancel("", BasicCancelOptions::default()).await;
                    }
                    return Ok(());
                }

                Some(delivery) = consumer.next() => {
                    match delivery {
                        Ok(d) => {
                            self.process_message(d.data).await;
                            d.ack(BasicAckOptions::default()).await;
                        }
                        Err(e) => {
                            error!(message error);
                        }
                    }
                }

	        _ = sleep(Duration::from_millis(500)) => {}
            }
        }
    }
}

I'm not entirely sure how this issue occurs. It seems similar to issue #357, but since I don't have deep experience with async Rust, I might be missing something obvious. Could it be that my implementation is incorrect?

Any guidance would be greatly appreciated!

[Keruspe]

Hello,

I guess it happens when you take more than 30min (1800000ms) to ack a message

[Keruspe]

Note that this is an error returned by the server that lapin just forwards, it doesn't seem to be linked to any thing in lapin?

[ohno104]

Hi @Keruspe,

I really appreciate your valuable insights—thank you!

I guess it happens when you take more than 30min (1800000ms) to ack a message

• I think my processing flow shouldn’t take more than 30 minutes, but I’ll double-check to be sure.

Note that this is an error returned by the server that lapin just forwards, it doesn't seem to be linked to any thing in lapin?

• yes, lapin just forwards the error via lapin-2.5.0/src/channel.rs:890, and it seems like the channel is automatically closing even though my connection remains active.

Do you think there’s anything in my implementation that might be causing this, or is there something I could be missing? If I’m not exceeding the 30-minute (1800000ms) ack limit, I wonder what else might trigger this behavior.

Thanks again for your time and help—I really appreciate it!

[Keruspe]

After how long does this usually happen, and how much messages do you deal with per second/minute/hour (depending on your scale) ?