From 4cb86c7dd4ba8dababf4a4d396c35fd6931239d4 Mon Sep 17 00:00:00 2001
From: Wez Furlong <wez@wezfurlong.org>
Date: Tue, 20 Jun 2023 07:11:22 -0700
Subject: [PATCH] queue: centralize insert_ready logic

Commit f3be8727913298646c5ab125832ce961a7a0d4aa was only an issue
because we had essentially two copies of the logic to insert
into the ready queue.

This commit refactors the second location to work in terms of the
first.

In doing so, I think this also fixes an issue with queues configured
for non-SMTP delivery, such as lua delivery implementations for
webhooks/AMQP in the delayed->ready case.  I think this explains
some weirdness that Tom has seen in the past when playing with
webhook logging.
---
 crates/kumod/src/queue.rs | 74 +++++++--------------------------------
 1 file changed, 13 insertions(+), 61 deletions(-)

diff --git a/crates/kumod/src/queue.rs b/crates/kumod/src/queue.rs
index 300edd31..6bb75d19 100644
--- a/crates/kumod/src/queue.rs
+++ b/crates/kumod/src/queue.rs
@@ -810,76 +810,28 @@ async fn maintain_named_queue(queue: &QueueHandle) -> anyhow::Result<()> {
                 continue;
             }
 
-            let now = Utc::now();
             match q.queue.pop() {
                 PopResult::Items(messages) => {
                     q.delayed_gauge.sub(messages.len() as i64);
-                    let max_age = q.queue_config.get_max_age();
                     tracing::trace!("{} msgs are now ready", messages.len());
 
                     for msg in messages {
-                        let egress_source =
-                            q.rr.next().ok_or_else(|| anyhow!("no sources in pool"))?;
+                        let msg = (*msg).clone();
 
-                        match ReadyQueueManager::resolve_by_queue_name(
-                            &q.name,
-                            &q.queue_config,
-                            &egress_source,
-                            &q.rr.name,
-                        )
-                        .await
-                        {
-                            Ok(site) => {
-                                let mut site = site.lock().await;
+                        if let Err(err) = q.insert_ready(msg.clone()).await {
+                            tracing::debug!("insert_ready: {err:#}");
 
-                                let msg = (*msg).clone();
-                                let id = *msg.id();
-
-                                let age = msg.age(now);
-                                if age >= max_age {
-                                    // TODO: log failure due to expiration
-                                    tracing::debug!("expiring {id} {age} > {max_age}");
-                                    SpoolManager::remove_from_spool(id).await?;
-                                    continue;
+                            if err.downcast_ref::<ReadyQueueFull>().is_none() {
+                                // It was a legit error while trying to do something useful
+                                match q.increment_attempts_and_update_delay(msg).await? {
+                                    Some(msg) => {
+                                        q.force_into_delayed(msg).await?;
+                                    }
+                                    None => {}
                                 }
-
-                                match site.insert(msg.clone()).await {
-                                    Ok(_) => {}
-                                    Err(_) => loop {
-                                        msg.delay_with_jitter(60).await?;
-                                        if matches!(
-                                            q.insert_delayed(msg.clone()).await?,
-                                            InsertResult::Delayed
-                                        ) {
-                                            break;
-                                        }
-                                    },
-                                }
-                            }
-                            Err(err) => {
-                                tracing::error!("Failed to resolve {}: {err:#}", q.name);
-                                log_disposition(LogDisposition {
-                                    kind: RecordType::TransientFailure,
-                                    msg: (*msg).clone(),
-                                    site: "",
-                                    peer_address: None,
-                                    response: Response {
-                                        code: 451,
-                                        enhanced_code: Some(EnhancedStatusCode {
-                                            class: 4,
-                                            subject: 4,
-                                            detail: 4,
-                                        }),
-                                        content: format!("failed to resolve {}: {err:#}", q.name),
-                                        command: None,
-                                    },
-                                    egress_pool: None,
-                                    egress_source: None,
-                                    relay_disposition: None,
-                                    delivery_protocol: None,
-                                })
-                                .await;
-                                q.requeue_message((*msg).clone(), true, None).await?;
+                            } else {
+                                // Queue is full; try again shortly
+                                q.force_into_delayed(msg).await?;
                             }
                         }
                     }