Protect UpdateWorkflowExecution API with dynamic config feature flag (#3884)

Author: alexshtin

common/dynamicconfig/constants.go

@@ -253,6 +253,11 @@ const (
 	// FrontendEnableBatcher enables batcher-related RPCs in the frontend
 	FrontendEnableBatcher = "frontend.enableBatcher"
 
+	// FrontendEnableUpdateWorkflowExecution enables UpdateWorkflowExecution API in the frontend.
+	//  UpdateWorkflowExecution API is under active development and is not ready for production use.
+	//  Default value is `false`. It will be changed to `true` when this API is ready and fully tested.
+	FrontendEnableUpdateWorkflowExecution = "frontend.enableUpdateWorkflowExecution"
+
 	// DeleteNamespaceDeleteActivityRPS is an RPS per every parallel delete executions activity.
 	// Total RPS is equal to DeleteNamespaceDeleteActivityRPS * DeleteNamespaceConcurrentDeleteExecutionsActivities.
 	// Default value is 100.

service/frontend/errors.go

@@ -110,4 +110,6 @@ var (
 	errBatchOpsWorkflowFilterNotSet      = serviceerror.NewInvalidArgument("Workflow executions and visibility filter are not set on request.")
 	errBatchOpsWorkflowFiltersNotAllowed = serviceerror.NewInvalidArgument("Workflow executions and visibility filter are both set on request. Only one of them is allowed.")
 	errBatchOpsMaxWorkflowExecutionCount = serviceerror.NewInvalidArgument("Workflow executions count exceeded.")
+
+	errUpdateWorkflowExecutionAPINotAllowed = serviceerror.NewPermissionDenied("UpdateWorkflowExecution operation is disabled on this namespace.", "")
 )

service/frontend/service.go

@@ -165,6 +165,8 @@ type Config struct {
 	// Batch operation dynamic configs
 	MaxConcurrentBatchOperation     dynamicconfig.IntPropertyFnWithNamespaceFilter
 	MaxExecutionCountBatchOperation dynamicconfig.IntPropertyFnWithNamespaceFilter
+
+	EnableUpdateWorkflowExecution dynamicconfig.BoolPropertyFnWithNamespaceFilter
 }
 
 // NewConfig returns new service config with default values
@@ -235,6 +237,8 @@ func NewConfig(dc *dynamicconfig.Collection, numHistoryShards int32, enableReadF
 		EnableBatcher:                   dc.GetBoolPropertyFnWithNamespaceFilter(dynamicconfig.FrontendEnableBatcher, true),
 		MaxConcurrentBatchOperation:     dc.GetIntPropertyFilteredByNamespace(dynamicconfig.FrontendMaxConcurrentBatchOperationPerNamespace, 1),
 		MaxExecutionCountBatchOperation: dc.GetIntPropertyFilteredByNamespace(dynamicconfig.FrontendMaxExecutionCountBatchOperationPerNamespace, 1000),
+
+		EnableUpdateWorkflowExecution: dc.GetBoolPropertyFnWithNamespaceFilter(dynamicconfig.FrontendEnableUpdateWorkflowExecution, false),
 	}
 }
 

service/frontend/workflow_handler.go

@@ -3684,6 +3684,10 @@ func (wh *WorkflowHandler) UpdateWorkflowExecution(
 		return nil, err
 	}
 
+	if !wh.config.EnableUpdateWorkflowExecution(request.Namespace) {
+		return nil, errUpdateWorkflowExecutionAPINotAllowed
+	}
+
 	histResp, err := wh.historyClient.UpdateWorkflowExecution(ctx, &historyservice.UpdateWorkflowExecutionRequest{
 		NamespaceId: nsID.String(),
 		Request:     request,

tests/dynamicconfig.go

@@ -52,6 +52,7 @@ var (
 		dynamicconfig.ReplicationTaskProcessorErrorRetryWait:         time.Millisecond,
 		dynamicconfig.ClusterMetadataRefreshInterval:                 100 * time.Millisecond,
 		dynamicconfig.NamespaceCacheRefreshInterval:                  NamespaceCacheRefreshInterval,
+		dynamicconfig.FrontendEnableUpdateWorkflowExecution:          true,
 	}
 )
 

tests/update_workflow_test.go

@@ -208,6 +208,7 @@ func (s *integrationSuite) TestUpdateWorkflow_FirstWorkflowTask_AcceptComplete()
 						Input: &updatepb.Input{Name: "update_handler", Args: payloads.EncodeString("update args")},
 					},
 				})
+				s.NoError(err1)
 				updateResultCh <- UpdateResult{Response: updateResponse, Err: err1}
 			}
 			go updateWorkflowFn()
@@ -402,6 +403,7 @@ func (s *integrationSuite) TestUpdateWorkflow_NewWorkflowTask_AcceptComplete() {
 				},
 			},
 		})
+		s.NoError(err1)
 		updateResultCh <- UpdateResult{Response: updateResponse, Err: err1}
 	}
 	go updateWorkflowFn()
@@ -560,6 +562,7 @@ func (s *integrationSuite) TestUpdateWorkflow_FirstWorkflowTask_Reject() {
 				},
 			},
 		})
+		s.NoError(err1)
 		updateResultCh <- UpdateResult{Response: updateResponse, Err: err1}
 	}
 	go updateWorkflowFn()
@@ -732,6 +735,7 @@ func (s *integrationSuite) TestUpdateWorkflow_NewWorkflowTask_Reject() {
 				},
 			},
 		})
+		s.NoError(err1)
 		updateResultCh <- UpdateResult{Response: updateResponse, Err: err1}
 	}
 	go updateWorkflowFn()
@@ -969,6 +973,7 @@ func (s *integrationSuite) TestUpdateWorkflow_FirstWorkflowTask_1stAccept_2ndAcc
 				},
 			},
 		})
+		s.NoError(err1)
 		ch <- UpdateResult{Response: updateResponse, Err: err1}
 	}
 
@@ -1246,6 +1251,7 @@ func (s *integrationSuite) TestUpdateWorkflow_FirstWorkflowTask_1stAccept_2ndRej
 				},
 			},
 		})
+		s.NoError(err1)
 		ch <- UpdateResult{Response: updateResponse, Err: err1}
 	}