Distributed SMTP queues and Rate limiting passing tests

tests/src/smtp/queue/manager.rs

@@ -21,49 +21,57 @@
  * for more details.
 */
 
-use std::time::{Duration, Instant};
+use std::{sync::Arc, time::Duration};
 
 use mail_auth::hickory_resolver::proto::op::ResponseCode;
 
-use smtp::queue::{manager::Queue, Domain, Message, Schedule, Status};
+use smtp::{
+    core::SMTP,
+    queue::{Domain, Message, Schedule, Status},
+};
+use store::write::now;
 
-#[test]
-fn queue_due() {
-    let mut queue = Queue::default();
+use crate::smtp::{TestConfig, TestSMTP};
+
+#[tokio::test]
+async fn queue_due() {
+    let mut core = SMTP::test();
+    let qr = core.init_test_queue("smtp_queue_due_test");
+    let core = Arc::new(core);
 
     let mut message = new_message(0);
     message.domains.push(domain("c", 3, 8, 9));
-    queue.schedule(Schedule {
-        due: message.next_delivery_event(),
-        inner: message,
-    });
+    let due = message.next_delivery_event();
+    message.save_changes(&core, 0.into(), due.into()).await;
 
     let mut message = new_message(1);
     message.domains.push(domain("b", 2, 6, 7));
-    queue.schedule(Schedule {
-        due: message.next_delivery_event(),
-        inner: message,
-    });
+    let due = message.next_delivery_event();
+    message.save_changes(&core, 0.into(), due.into()).await;
 
     let mut message = new_message(2);
     message.domains.push(domain("a", 1, 4, 5));
-    queue.schedule(Schedule {
-        due: message.next_delivery_event(),
-        inner: message,
-    });
+    let due = message.next_delivery_event();
+    message.save_changes(&core, 0.into(), due.into()).await;
 
     for domain in vec!["a", "b", "c"].into_iter() {
-        let wake_up = queue.wake_up_time();
-        assert!(
-            (900..=1000).contains(&wake_up.as_millis()),
-            "{}",
-            wake_up.as_millis()
-        );
-        std::thread::sleep(wake_up);
-        queue.next_due().unwrap().domain(domain);
+        let now = now();
+        for queue_event in core.next_event().await {
+            if queue_event.due > now {
+                let wake_up = queue_event.due - now;
+                assert_eq!(wake_up, 1);
+                std::thread::sleep(Duration::from_secs(wake_up));
+            }
+            if let Some(message) = core.read_message(queue_event.queue_id).await {
+                message.domain(domain);
+                message.remove(&core, queue_event.due).await;
+            } else {
+                panic!("Message not found");
+            }
+        }
     }
 
-    assert!(queue.next_due().is_none());
+    qr.assert_queue_is_empty().await;
 }
 
 #[test]
@@ -127,11 +135,10 @@ fn delivery_events() {
     assert!(message.next_event().is_none());
 }
 
-pub fn new_message(id: u64) -> Box<Message> {
-    Box::new(Message {
+pub fn new_message(id: u64) -> Message {
+    Message {
         size: 0,
         id,
-        path: Default::default(),
         created: 0,
         return_path: "sender@foobar.org".to_string(),
         return_path_lcase: "".to_string(),
@@ -141,8 +148,9 @@ pub fn new_message(id: u64) -> Box<Message> {
         flags: 0,
         env_id: None,
         priority: 0,
-        queue_refs: vec![],
-    })
+        quota_keys: vec![],
+        blob_hash: Default::default(),
+    }
 }
 
 fn domain(domain: &str, retry: u64, notify: u64, expires: u64) -> Domain {
@@ -150,10 +158,9 @@ fn domain(domain: &str, retry: u64, notify: u64, expires: u64) -> Domain {
         domain: domain.to_string(),
         retry: Schedule::later(Duration::from_secs(retry)),
         notify: Schedule::later(Duration::from_secs(notify)),
-        expires: Instant::now() + Duration::from_secs(expires),
+        expires: now() + expires,
         status: Status::Scheduled,
         disable_tls: false,
-        changed: false,
     }
 }
 
@@ -164,7 +171,10 @@ pub trait TestMessage {
 
 impl TestMessage for Message {
     fn domain(&self, name: &str) -> &Domain {
-        self.domains.iter().find(|d| d.domain == name).unwrap()
+        self.domains
+            .iter()
+            .find(|d| d.domain == name)
+            .unwrap_or_else(|| panic!("Expected domain {name} not found in {:?}", self.domains))
     }
 
     fn domain_mut(&mut self, name: &str) -> &mut Domain {