(nPartitions, nWorkers int, timeToDrain time.Duration, useNewMatching bool)
| 73 | } |
| 74 | |
| 75 | func (s *TaskQueueSuite) taskQueueRateLimitTest(nPartitions, nWorkers int, timeToDrain time.Duration, useNewMatching bool) { |
| 76 | env := s.newTestEnv( |
| 77 | testcore.WithDynamicConfig(dynamicconfig.MatchingUseNewMatcher, useNewMatching), |
| 78 | testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, nPartitions), |
| 79 | testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, nPartitions), |
| 80 | // exclude the effect of the default forwarding rate limit (10) |
| 81 | testcore.WithDynamicConfig(dynamicconfig.MatchingForwarderMaxRatePerSecond, 1000), |
| 82 | // 30 tasks at 1 task per second is 30 seconds. |
| 83 | // if invalid tasks are NOT using the rate limit, then this should take well below that long. |
| 84 | // task forwarding between task queue partitions is rate-limited by default to 10 rps. |
| 85 | testcore.WithDynamicConfig(dynamicconfig.AdminMatchingNamespaceTaskqueueToPartitionDispatchRate, 1), |
| 86 | testcore.WithDynamicConfig(dynamicconfig.TaskQueueInfoByBuildIdTTL, 0), |
| 87 | // Terminating workflows mid-backlog intentionally trips soft asserts in the |
| 88 | // matching task queue; disable fail-on-error so those logs don't poison the |
| 89 | // suite's shared cluster. |
| 90 | testcore.WithDisableTestloggerFailure(), |
| 91 | ) |
| 92 | |
| 93 | const maxBacklog = 30 |
| 94 | tv := testvars.New(s.T()) |
| 95 | |
| 96 | helloRateLimitTest := func(ctx workflow.Context, name string) (string, error) { |
| 97 | return "Hello " + name + " !", nil |
| 98 | } |
| 99 | |
| 100 | // start workflows to create a backlog |
| 101 | for wfidx := range maxBacklog { |
| 102 | _, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{ |
| 103 | TaskQueue: tv.TaskQueue().GetName(), |
| 104 | ID: fmt.Sprintf("wf%d", wfidx), |
| 105 | }, helloRateLimitTest, "Donna") |
| 106 | s.NoError(err) |
| 107 | } |
| 108 | |
| 109 | // wait for backlog to be >= maxBacklog |
| 110 | wfBacklogCount := int64(0) |
| 111 | s.Eventually( |
| 112 | func() bool { |
| 113 | wfBacklogCount = s.getBacklogCount(env, tv) |
| 114 | return wfBacklogCount >= maxBacklog |
| 115 | }, |
| 116 | 5*time.Second, |
| 117 | 200*time.Millisecond, |
| 118 | ) |
| 119 | |
| 120 | // terminate all those workflow executions so that all the tasks in the backlog are invalid |
| 121 | var wfList []*workflowpb.WorkflowExecutionInfo |
| 122 | s.Eventually( |
| 123 | func() bool { |
| 124 | listResp, err := env.FrontendClient().ListWorkflowExecutions(s.Context(), &workflowservice.ListWorkflowExecutionsRequest{ |
| 125 | Namespace: env.Namespace().String(), |
| 126 | Query: fmt.Sprintf("TaskQueue = '%s'", tv.TaskQueue().GetName()), |
| 127 | }) |
| 128 | s.NoError(err) |
| 129 | wfList = listResp.GetExecutions() |
| 130 | return len(wfList) == maxBacklog |
| 131 | }, |
| 132 | 5*time.Second, |
no test coverage detected