[Visibility] Read Chasm Workflows

Author: rodrigozhou

common/dynamicconfig/constants.go

@@ -3110,6 +3110,13 @@ the CHASM implementation. When disabled, new update callbacks will not be regist
 but existing callbacks will still be processed and fired.`,
 	)
 
+	EnableReadChasmWorkflows = NewNamespaceBoolSetting(
+		"frontend.enableReadChasmWorkflows",
+		false,
+		`If enabled, Visibility read workflow APIS (ListWorkflowExecutions, CountWorkflowsExecutions)
+will use the CHASM API in VisibilityManager`,
+	)
+
 	VersionMembershipCacheTTL = NewGlobalDurationSetting(
 		"history.versionMembershipCacheTTL",
 		1*time.Second,

common/persistence/visibility/visibility_manager_impl.go

@@ -174,6 +174,7 @@ func (p *visibilityManagerImpl) ListChasmExecutions(
 		executionInfo, err := p.convertToChasmExecutionInfo(
 			exec,
 			mapper,
+			request.ArchetypeId,
 			namespace.Name(request.Namespace),
 		)
 		if err != nil {
@@ -192,6 +193,7 @@ func (p *visibilityManagerImpl) ListChasmExecutions(
 func (p *visibilityManagerImpl) convertToChasmExecutionInfo(
 	exec *store.InternalExecutionInfo,
 	mapper *chasm.VisibilitySearchAttributesMapper,
+	archetypeID chasm.ArchetypeID,
 	namespaceName namespace.Name,
 ) (*chasmspb.VisibilityExecutionInfo, error) {
 	customSAs, chasmSAs := splitSearchAttributes(exec.SearchAttributes)
@@ -212,6 +214,48 @@ func (p *visibilityManagerImpl) convertToChasmExecutionInfo(
 			enumspb.INDEXED_VALUE_TYPE_KEYWORD,
 		)
 	}
+	if archetypeID == chasm.WorkflowArchetypeID {
+		if !exec.ExecutionTime.After(time.Unix(0, 0)) {
+			exec.ExecutionTime = exec.StartTime
+		}
+
+		chasmAliasedSAs.IndexedFields[sadefs.ExecutionStatus] = sadefs.MustEncodeValue(
+			exec.Status.String(),
+			enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		)
+		chasmAliasedSAs.IndexedFields[sadefs.WorkflowType] = sadefs.MustEncodeValue(
+			exec.TypeName,
+			enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		)
+		chasmAliasedSAs.IndexedFields[sadefs.ExecutionTime] = sadefs.MustEncodeValue(
+			exec.ExecutionTime,
+			enumspb.INDEXED_VALUE_TYPE_DATETIME,
+		)
+		if exec.ExecutionDuration > 0 {
+			chasmAliasedSAs.IndexedFields[sadefs.ExecutionDuration] = sadefs.MustEncodeValue(
+				exec.ExecutionDuration.Nanoseconds(),
+				enumspb.INDEXED_VALUE_TYPE_INT,
+			)
+		}
+		if exec.ParentWorkflowID != "" {
+			chasmAliasedSAs.IndexedFields[sadefs.ParentWorkflowID] = sadefs.MustEncodeValue(
+				exec.ParentWorkflowID,
+				enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+			)
+			chasmAliasedSAs.IndexedFields[sadefs.ParentRunID] = sadefs.MustEncodeValue(
+				exec.ParentRunID,
+				enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+			)
+		}
+		chasmAliasedSAs.IndexedFields[sadefs.RootWorkflowID] = sadefs.MustEncodeValue(
+			exec.RootWorkflowID,
+			enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		)
+		chasmAliasedSAs.IndexedFields[sadefs.RootRunID] = sadefs.MustEncodeValue(
+			exec.RootRunID,
+			enumspb.INDEXED_VALUE_TYPE_KEYWORD,
+		)
+	}
 
 	customAliasedSAs, err := searchattribute.AliasFields(
 		p.searchAttributesMapperProvider,

common/searchattribute/sadefs/encode_value.go

@@ -33,6 +33,17 @@ func MustEncodeValue(val any, t enumspb.IndexedValueType) *commonpb.Payload {
 	return valPayload
 }
 
+// MustDecodeValue decodes search attribute Payload and IndexedValueType to value.
+// Panics if it fails to decode.
+func MustDecodeValue(value *commonpb.Payload, t enumspb.IndexedValueType) any {
+	v, err := DecodeValue(value, t, false)
+	if err != nil {
+		// nolint:forbidigo
+		panic(err)
+	}
+	return v
+}
+
 // DecodeValue decodes search attribute value from Payload using (in order):
 // 1. passed type t.
 // 2. type from MetadataType field, if t is not specified.

common/searchattribute/sadefs/encode_value_test.go

@@ -392,6 +392,72 @@ func Test_MustEncodeValue(t *testing.T) {
 	})
 }
 
+func Test_MustDecodeValue(t *testing.T) {
+	t.Run("success", func(t *testing.T) {
+		encodedValue, err := payload.Encode("foo")
+		require.NoError(t, err)
+		decodedValue := MustDecodeValue(encodedValue, enumspb.INDEXED_VALUE_TYPE_KEYWORD)
+		require.Equal(t, "foo", decodedValue)
+	})
+
+	t.Run("success with nil value", func(t *testing.T) {
+		encodedValue, err := payload.Encode(nil)
+		require.NoError(t, err)
+		decodedValue := MustDecodeValue(encodedValue, enumspb.INDEXED_VALUE_TYPE_KEYWORD)
+		require.Nil(t, decodedValue)
+	})
+
+	t.Run("success with metadata type", func(t *testing.T) {
+		encodedValue, err := payload.Encode(int64(123))
+		require.NoError(t, err)
+		encodedValue.Metadata[MetadataType] = []byte(enumspb.INDEXED_VALUE_TYPE_INT.String())
+		decodedValue := MustDecodeValue(encodedValue, enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED)
+		require.Equal(t, int64(123), decodedValue)
+	})
+
+	t.Run("panic on unspecified type without metadata", func(t *testing.T) {
+		defer func() {
+			r := recover()
+			require.NotNil(t, r)
+			err, ok := r.(error)
+			require.True(t, ok)
+			require.ErrorIs(t, err, ErrInvalidType)
+		}()
+		encodedValue, err := payload.Encode(int64(123))
+		require.NoError(t, err)
+		_ = MustDecodeValue(encodedValue, enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED)
+		require.Fail(t, "MustDecodeValue did not panic with unspecified type")
+	})
+
+	t.Run("panic on type mismatch", func(t *testing.T) {
+		defer func() {
+			r := recover()
+			require.NotNil(t, r)
+			err, ok := r.(error)
+			require.True(t, ok)
+			require.ErrorIs(t, err, converter.ErrUnableToDecode)
+		}()
+		encodedValue, err := payload.Encode(int64(123))
+		require.NoError(t, err)
+		_ = MustDecodeValue(encodedValue, enumspb.INDEXED_VALUE_TYPE_KEYWORD)
+		require.Fail(t, "MustDecodeValue did not panic with type mismatch")
+	})
+
+	t.Run("panic on list not allowed", func(t *testing.T) {
+		defer func() {
+			r := recover()
+			require.NotNil(t, r)
+			err, ok := r.(error)
+			require.True(t, ok)
+			require.ErrorIs(t, err, converter.ErrUnableToDecode)
+		}()
+		encodedValue, err := payload.Encode([]int64{123, 456})
+		require.NoError(t, err)
+		_ = MustDecodeValue(encodedValue, enumspb.INDEXED_VALUE_TYPE_INT)
+		require.Fail(t, "MustDecodeValue did not panic with list of values")
+	})
+}
+
 func Test_ValidateStrings(t *testing.T) {
 	_, err := validateStrings("anything here", errors.New("test error"))
 	assert.Error(t, err)

service/frontend/service.go

@@ -178,6 +178,8 @@ type Config struct {
 	// Enables ID-space collision sentinels, and must be enabled and propagated in
 	// advance of EnableCHASMSchedulerCreation.
 	EnableCHASMSchedulerSentinels dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	// Enable ListChasmExecutions to list workflows.
+	EnableReadChasmWorkflows dynamicconfig.BoolPropertyFnWithNamespaceFilter
 
 	// Enable deployment RPCs
 	EnableDeployments dynamicconfig.BoolPropertyFnWithNamespaceFilter
@@ -363,6 +365,7 @@ func NewConfig(
 		CHASMSchedulerCreationRolloutPercent: dynamicconfig.CHASMSchedulerCreationRolloutPercent.Get(dc),
 		EnableCHASMSchedulerRouting:          dynamicconfig.EnableCHASMSchedulerRouting.Get(dc),
 		EnableCHASMSchedulerSentinels:        dynamicconfig.EnableCHASMSchedulerSentinels.Get(dc),
+		EnableReadChasmWorkflows:             dynamicconfig.EnableReadChasmWorkflows.Get(dc),
 
 		// [cleanup-wv-pre-release]
 		EnableDeployments:        dynamicconfig.EnableDeployments.Get(dc),