Test with more values

Author: Sushisource

core/src/worker/workflow/wft_poller.rs

@@ -104,11 +104,6 @@ impl WFTPollerShared {
         if is_sticky {
             self.wait_for_first_nonsticky_poll.notified().await;
         }
-        info!(
-            "Waiting if needed. Sticky: {} / backlog {}",
-            is_sticky,
-            *self.last_seen_sticky_backlog.0.borrow()
-        );
         // If there's a sticky backlog, prioritize it.
         if !is_sticky {
             let backlog = *self.last_seen_sticky_backlog.0.borrow();
@@ -127,12 +122,6 @@ impl WFTPollerShared {
             if let Some((sticky_active, non_sticky_active)) =
                 self.sticky_active.get().zip(self.non_sticky_active.get())
             {
-                info!(
-                    "Balance (sticky {}), non-sticky {} sticky {}",
-                    is_sticky,
-                    *non_sticky_active.borrow(),
-                    *sticky_active.borrow()
-                );
                 if is_sticky {
                     let _ = sticky_active
                         .clone()
@@ -147,7 +136,6 @@ impl WFTPollerShared {
                         })
                         .await;
                 }
-                info!("Done balance (sticky {})", is_sticky);
             }
         }
     }

tests/integ_tests/polling_tests.rs

@@ -7,12 +7,15 @@ use temporal_sdk_core::{
     ClientOptionsBuilder, ephemeral_server::TemporalDevServerConfigBuilder, init_worker,
 };
 use temporal_sdk_core_api::{Worker, worker::PollerBehavior};
-use temporal_sdk_core_protos::coresdk::{
-    AsJsonPayloadExt, IntoCompletion,
-    activity_task::activity_task as act_task,
-    workflow_activation::{FireTimer, WorkflowActivationJob, workflow_activation_job},
-    workflow_commands::{ActivityCancellationType, RequestCancelActivity, StartTimer},
-    workflow_completion::WorkflowActivationCompletion,
+use temporal_sdk_core_protos::{
+    coresdk::{
+        AsJsonPayloadExt, IntoCompletion,
+        activity_task::activity_task as act_task,
+        workflow_activation::{FireTimer, WorkflowActivationJob, workflow_activation_job},
+        workflow_commands::{ActivityCancellationType, RequestCancelActivity, StartTimer},
+        workflow_completion::WorkflowActivationCompletion,
+    },
+    temporal::api::enums::v1::EventType,
 };
 use temporal_sdk_core_test_utils::{
     CoreWfStarter, WorkerTestHelpers, default_cached_download, drain_pollers_and_shutdown,
@@ -209,16 +212,20 @@ async fn switching_worker_client_changes_poll() {
     server2.shutdown().await.unwrap();
 }
 
+#[rstest::rstest]
 #[tokio::test]
-async fn only_one_workflow_slot_and_two_pollers() {
+async fn small_workflow_slots_and_pollers(
+    #[values(1, 2)] wft_slots: usize,
+    #[values(1, 2)] wft_pollers: usize,
+) {
     let wf_name = "only_one_workflow_slot_and_two_pollers";
     let mut starter = CoreWfStarter::new(wf_name);
     starter
         .worker_config
-        .max_outstanding_workflow_tasks(2_usize)
+        .max_outstanding_workflow_tasks(wft_slots)
         .max_outstanding_local_activities(1_usize)
         .activity_task_poller_behavior(PollerBehavior::SimpleMaximum(1))
-        .workflow_task_poller_behavior(PollerBehavior::SimpleMaximum(2))
+        .workflow_task_poller_behavior(PollerBehavior::SimpleMaximum(wft_pollers))
         .max_outstanding_activities(1_usize);
     let mut worker = starter.worker().await;
     worker.register_wf(wf_name.to_owned(), |ctx: WfContext| async move {
@@ -236,7 +243,7 @@ async fn only_one_workflow_slot_and_two_pollers() {
     worker.register_activity("echo_activity", echo);
     worker
         .submit_wf(
-            wf_name.to_string(),
+            starter.get_task_queue(),
             wf_name.to_owned(),
             vec![],
             WorkflowOptions::default(),
@@ -245,4 +252,11 @@ async fn only_one_workflow_slot_and_two_pollers() {
         .unwrap();
     // If we don't fail the workflow on nondeterminism, we'll get stuck here retrying the WFT
     worker.run_until_done().await.unwrap();
+    // Verify no task timeouts happened
+    let history = starter.get_history().await;
+    let any_task_timeouts = history
+        .events
+        .iter()
+        .any(|e| e.event_type() == EventType::WorkflowTaskTimedOut);
+    assert!(!any_task_timeouts);
 }