Validate Priority metadata (#8159)

## What changed?
Add validation for Priority metadata.

## Why?
Negative keys/weights are ignored but we should reject requests with
too-long keys.

## How did you test it?
- [x] added new unit test(s)

Author: dnr

common/priorities/priority_util.go

@@ -4,6 +4,17 @@ import (
 	"cmp"
 
 	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/api/serviceerror"
+)
+
+const (
+	fairnessKeyMaxLength = 64
+)
+
+var (
+	ErrInvalidPriority       = serviceerror.NewInvalidArgument("PriorityKey can't be negative")
+	ErrFairnessKeyLength     = serviceerror.NewInvalidArgument("FairnessKey length exceeds limit")
+	ErrInvalidFairnessWeight = serviceerror.NewInvalidArgument("FairnessWeight can't be negative")
 )
 
 func Merge(
@@ -22,3 +33,16 @@ func Merge(
 		FairnessWeight: cmp.Or(override.FairnessWeight, base.FairnessWeight),
 	}
 }
+
+func Validate(p *commonpb.Priority) error {
+	if p == nil {
+		return nil
+	} else if p.PriorityKey < 0 {
+		return ErrInvalidPriority
+	} else if len(p.FairnessKey) > fairnessKeyMaxLength {
+		return ErrFairnessKeyLength
+	} else if p.FairnessWeight < 0 {
+		return ErrInvalidFairnessWeight
+	}
+	return nil
+}

common/priorities/priority_util_test.go

@@ -1,8 +1,10 @@
 package priorities
 
 import (
+	"strings"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	commonpb "go.temporal.io/api/common/v1"
 )
@@ -79,3 +81,30 @@ func TestMerge(t *testing.T) {
 		})
 	}
 }
+
+func TestValidate(t *testing.T) {
+	testcases := []struct {
+		p   *commonpb.Priority
+		err bool
+	}{
+		{p: &commonpb.Priority{}},
+		{p: &commonpb.Priority{PriorityKey: 5}},
+		{p: &commonpb.Priority{PriorityKey: -5}, err: true},
+		{p: &commonpb.Priority{FairnessKey: "abcdefg"}},
+		{p: &commonpb.Priority{FairnessKey: strings.Repeat("abcdefg", 10)}, err: true},
+		{p: &commonpb.Priority{FairnessWeight: 0.1}},
+		{p: &commonpb.Priority{FairnessWeight: 1e10}},
+		{p: &commonpb.Priority{FairnessWeight: -3}, err: true},
+	}
+
+	for _, tc := range testcases {
+		t.Run("test", func(t *testing.T) {
+			err := Validate(tc.p)
+			if tc.err {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

service/frontend/workflow_handler.go

@@ -59,6 +59,7 @@ import (
 	"go.temporal.io/server/common/persistence/visibility/manager"
 	"go.temporal.io/server/common/primitives"
 	"go.temporal.io/server/common/primitives/timestamp"
+	"go.temporal.io/server/common/priorities"
 	"go.temporal.io/server/common/retrypolicy"
 	"go.temporal.io/server/common/rpc"
 	"go.temporal.io/server/common/rpc/interceptor"
@@ -487,6 +488,10 @@ func (wh *WorkflowHandler) prepareStartWorkflowRequest(
 		request.SearchAttributes = sa
 	}
 
+	if err := priorities.Validate(request.Priority); err != nil {
+		return nil, err
+	}
+
 	if err := wh.validateWorkflowCompletionCallbacks(namespaceName, request.GetCompletionCallbacks()); err != nil {
 		return nil, err
 	}
@@ -2065,6 +2070,10 @@ func (wh *WorkflowHandler) SignalWithStartWorkflowExecution(ctx context.Context,
 		request.SearchAttributes = sa
 	}
 
+	if err := priorities.Validate(request.Priority); err != nil {
+		return nil, err
+	}
+
 	if err := wh.validateLinks(namespaceName, request.GetLinks()); err != nil {
 		return nil, err
 	}

service/history/api/command_attr_validator.go

@@ -15,6 +15,7 @@ import (
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/primitives/timestamp"
+	"go.temporal.io/server/common/priorities"
 	"go.temporal.io/server/common/retrypolicy"
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/common/tqid"
@@ -130,6 +131,10 @@ func (v *CommandAttrValidator) ValidateActivityScheduleAttributes(
 		return failedCause, serviceerror.NewInvalidArgumentf("Invalid HeartbeatTimeout for ScheduleActivityTaskCommand: %v. ActivityId=%s ActivityType=%s", err, activityID, activityType)
 	}
 
+	if err := priorities.Validate(attributes.Priority); err != nil {
+		return failedCause, err
+	}
+
 	ScheduleToCloseSet := attributes.GetScheduleToCloseTimeout().AsDuration() > 0
 	ScheduleToStartSet := attributes.GetScheduleToStartTimeout().AsDuration() > 0
 	StartToCloseSet := attributes.GetStartToCloseTimeout().AsDuration() > 0
@@ -558,6 +563,10 @@ func (v *CommandAttrValidator) ValidateStartChildExecutionAttributes(
 		return enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_SEARCH_ATTRIBUTES, fmt.Errorf("invalid SearchAttributes on StartChildWorkflowCommand: %w. WorkflowId=%s WorkflowType=%s Namespace=%s", err, wfID, wfType, ns)
 	}
 
+	if err := priorities.Validate(attributes.Priority); err != nil {
+		return failedCause, err
+	}
+
 	// Inherit taskqueue from parent workflow execution if not provided on command
 	if attributes.TaskQueue == nil {
 		attributes.TaskQueue = &taskqueuepb.TaskQueue{