Add test

server/svix-server/src/cfg.rs

@@ -69,6 +69,10 @@ const DEFAULTS: &str = include_str!("../config.default.toml");
 
 pub type Configuration = Arc<ConfigurationInner>;
 
+fn default_redis_pending_duration_secs() -> u64 {
+    45
+}
+
 #[derive(Clone, Debug, Deserialize, Validate)]
 #[validate(
     schema(function = "validate_config_complete"),
@@ -200,6 +204,9 @@ pub struct ConfigurationInner {
     #[serde(flatten)]
     pub proxy_config: Option<ProxyConfig>,
 
+    #[serde(default = "default_redis_pending_duration_secs")]
+    pub redis_pending_duration_secs: u64,
+
     #[serde(flatten)]
     pub internal: InternalConfig,
 }

server/svix-server/src/queue/redis.rs

@@ -84,7 +84,7 @@ pub async fn new_pair(
 ) -> (TaskQueueProducer, TaskQueueConsumer) {
     new_pair_inner(
         cfg,
-        Duration::from_secs(45),
+        Duration::from_secs(cfg.redis_pending_duration_secs),
         prefix.unwrap_or_default(),
         MAIN,
         DELAYED,

server/svix-server/tests/it/redis_queue.rs

@@ -6,17 +6,27 @@
 
 use std::{str::FromStr, time::Duration};
 
+use http::StatusCode;
 use redis::AsyncCommands as _;
+use svix_ksuid::KsuidLike;
 use svix_server::{
     cfg::Configuration,
-    core::types::{ApplicationId, EndpointId, MessageAttemptTriggerType, MessageId},
+    core::types::{
+        ApplicationId, BaseId, EndpointId, MessageAttemptTriggerType, MessageId, OrganizationId,
+    },
     queue::{
         new_pair, MessageTask, QueueTask, TaskQueueConsumer, TaskQueueDelivery, TaskQueueProducer,
     },
     redis::RedisManager,
+    v1::endpoints::message::MessageOut,
 };
 use tokio::time::timeout;
 
+use crate::utils::{
+    common_calls::{create_test_app, create_test_endpoint, message_in},
+    get_default_test_config, start_svix_server_with_cfg_and_org_id_and_prefix,
+};
+
 // TODO: Don't copy this from the Redis queue test directly, place the fn somewhere both can access
 async fn get_pool(cfg: &Configuration) -> RedisManager {
     RedisManager::from_queue_backend(&cfg.queue_backend(), cfg.redis_pool_max_size).await
@@ -148,3 +158,90 @@ async fn test_many_queue_consumers_delayed() {
     )
     .await;
 }
+
+#[tokio::test]
+#[ignore]
+async fn test_redis_streams_dlq() {
+    let mut cfg = get_default_test_config();
+    cfg.worker_enabled = false;
+    cfg.redis_pending_duration_secs = 1;
+
+    let cfg = std::sync::Arc::new(cfg);
+    let prefix = svix_ksuid::Ksuid::new(None, None).to_string();
+
+    let pool = get_pool(&cfg).await;
+    let mut conn = pool.get().await.unwrap();
+
+    let _: () = conn
+        .del(format!("{prefix}{{queue}}_svix_v3_main"))
+        .await
+        .unwrap();
+
+    let _: () = conn
+        .del(format!("{prefix}{{queue}}_svix_dlq"))
+        .await
+        .unwrap();
+
+    let (client, _jh) = start_svix_server_with_cfg_and_org_id_and_prefix(
+        &cfg,
+        OrganizationId::new(None, None),
+        prefix.clone(),
+    )
+    .await;
+
+    let app_id = create_test_app(&client, "v1MessageCRTestApp")
+        .await
+        .unwrap()
+        .id;
+
+    let _endp_id = create_test_endpoint(&client, &app_id, "http://localhost:2/bad/url/")
+        .await
+        .unwrap()
+        .id;
+
+    let _message_1: MessageOut = client
+        .post(
+            &format!("api/v1/app/{app_id}/msg/"),
+            message_in(&app_id, serde_json::json!({"test": "value"})).unwrap(),
+            StatusCode::ACCEPTED,
+        )
+        .await
+        .unwrap();
+
+    let (_p, mut c) = new_pair(&cfg, Some(&prefix)).await;
+
+    let wait_time = std::time::Duration::from_millis(1_500);
+    for _ in 0..3 {
+        let res = c.receive_all(wait_time).await.unwrap();
+        assert!(!res.is_empty());
+        for j in res {
+            j.nack().await.unwrap();
+        }
+    }
+
+    let res = c.receive_all(wait_time).await.unwrap();
+    assert!(res.is_empty());
+
+    tokio::time::sleep(wait_time).await;
+
+    // Redrive
+    client
+        .put_without_response(
+            "/api/v1/admin/redrive-dlq",
+            serde_json::Value::Null,
+            StatusCode::NO_CONTENT,
+        )
+        .await
+        .unwrap();
+
+    for _ in 0..3 {
+        let res = c.receive_all(wait_time).await.unwrap();
+        assert!(!res.is_empty());
+        for j in res {
+            j.nack().await.unwrap();
+        }
+    }
+
+    let res = c.receive_all(wait_time).await.unwrap();
+    assert!(res.is_empty());
+}

server/svix-server/tests/it/utils/mod.rs

@@ -330,6 +330,15 @@ pub async fn start_svix_server_with_cfg(
 pub async fn start_svix_server_with_cfg_and_org_id(
     cfg: &ConfigurationInner,
     org_id: OrganizationId,
+) -> (TestClient, tokio::task::JoinHandle<()>) {
+    let prefix = svix_ksuid::Ksuid::new(None, None).to_string();
+    start_svix_server_with_cfg_and_org_id_and_prefix(cfg, org_id, prefix).await
+}
+
+pub async fn start_svix_server_with_cfg_and_org_id_and_prefix(
+    cfg: &ConfigurationInner,
+    org_id: OrganizationId,
+    prefix: String,
 ) -> (TestClient, tokio::task::JoinHandle<()>) {
     let (tracing_subscriber, _guard) = setup_tracing(cfg, /* for_test = */ true);
 
@@ -340,12 +349,8 @@ pub async fn start_svix_server_with_cfg_and_org_id(
     let base_uri = format!("http://{}", listener.local_addr().unwrap());
 
     let jh = tokio::spawn(
-        svix_server::run_with_prefix(
-            Some(svix_ksuid::Ksuid::new(None, None).to_string()),
-            cfg,
-            Some(listener),
-        )
-        .with_subscriber(tracing_subscriber),
+        svix_server::run_with_prefix(Some(prefix), cfg, Some(listener))
+            .with_subscriber(tracing_subscriber),
     );
 
     (TestClient::new(base_uri, &token), jh)