Separate Chasm ListExecutions interface, combine user and chasm memo, add support for legacy converters

Author: awln-temporal

chasm/engine.go

@@ -4,6 +4,11 @@ package chasm
 
 import (
 	"context"
+	"reflect"
+	"time"
+
+	commonpb "go.temporal.io/api/common/v1"
+	"google.golang.org/protobuf/proto"
 )
 
 // NoValue is a sentinel type representing no value.
@@ -46,6 +51,83 @@ type Engine interface {
 		func(MutableContext, Component, any) error,
 		...TransitionOption,
 	) ([]byte, error)
+
+	ListExecutions(
+		context.Context,
+		reflect.Type,
+		*ListChasmExecutionsRequest,
+		...ListChasmExecutionsOption,
+	) (*ListChasmExecutionsResponse, error)
+
+	CountExecutions(
+		context.Context,
+		reflect.Type,
+		*CountChasmExecutionsRequest,
+	) (*CountChasmExecutionsResponse, error)
+}
+
+type ChasmExecutionInfo struct {
+	BusinessID             string
+	RunID                  string
+	StartTime              time.Time
+	CloseTime              time.Time
+	HistoryLength          int64
+	HistorySizeBytes       int64
+	StateTransitionCount   int64
+	ChasmSearchAttributes  SearchAttributesMap
+	CustomSearchAttributes map[string]*commonpb.Payload
+	Memo                   *commonpb.Memo
+	ChasmMemo              *commonpb.Payload
+}
+
+type ListChasmExecutionsRequest struct {
+	NamespaceID   string
+	NamespaceName string
+	Query         string
+}
+
+type ListChasmExecutionsOptions struct {
+	PageSize      int
+	NextPageToken []byte
+}
+
+type ListChasmExecutionsOption func(*ListChasmExecutionsOptions)
+
+func WithPagination(
+	pageSize int,
+	nextPageToken []byte,
+) ListChasmExecutionsOption {
+	return func(req *ListChasmExecutionsOptions) {
+		req.PageSize = pageSize
+		req.NextPageToken = nextPageToken
+	}
+}
+
+type ListChasmExecutionsResponse struct {
+	Executions    []*ChasmExecutionInfo
+	NextPageToken []byte
+}
+
+type CountChasmExecutionsRequest struct {
+	NamespaceID   string
+	NamespaceName string
+	Query         string
+}
+
+type CountChasmExecutionsResponse struct {
+	Count int64
+}
+
+// TypedChasmRunInfo provides type-safe access to ChasmMemo
+type TypedChasmExecutionInfo[M proto.Message] struct {
+	*ChasmExecutionInfo
+	TypedChasmMemo M
+}
+
+// TypedListChasmRunsResponse provides type-safe response with unmarshaled memos
+type TypedListChasmExecutionsResponse[M proto.Message] struct {
+	Executions    []*TypedChasmExecutionInfo[M]
+	NextPageToken []byte
 }
 
 type BusinessIDReusePolicy int
@@ -302,3 +384,44 @@ func engineFromContext(
 	}
 	return e
 }
+
+func ListExecutions[C Component, M proto.Message](
+	ctx context.Context,
+	request *ListChasmExecutionsRequest,
+	opts ...ListChasmExecutionsOption,
+) (*TypedListChasmExecutionsResponse[M], error) {
+	archetypeType := reflect.TypeFor[C]()
+	response, err := engineFromContext(ctx).ListExecutions(ctx, archetypeType, request, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	// Convert response, unmarshaling ChasmMemo to type M
+	typedExecutions := make([]*TypedChasmExecutionInfo[M], len(response.Executions))
+	for i, execution := range response.Executions {
+		var typedMemo M
+		if len(execution.ChasmMemo.Data) > 0 {
+			msg := reflect.New(reflect.TypeFor[M]()).Interface().(M)
+			if err := proto.Unmarshal(execution.ChasmMemo.Data, msg); err == nil {
+				typedMemo = msg
+			}
+		}
+		typedExecutions[i] = &TypedChasmExecutionInfo[M]{
+			ChasmExecutionInfo: execution,
+			TypedChasmMemo:     typedMemo,
+		}
+	}
+
+	return &TypedListChasmExecutionsResponse[M]{
+		Executions:    typedExecutions,
+		NextPageToken: response.NextPageToken,
+	}, nil
+}
+
+func CountExecutions[C Component](
+	ctx context.Context,
+	request *CountChasmExecutionsRequest,
+) (*CountChasmExecutionsResponse, error) {
+	archetypeType := reflect.TypeFor[C]()
+	return engineFromContext(ctx).CountExecutions(ctx, archetypeType, request)
+}

chasm/engine_mock.go

@@ -14,10 +14,6 @@ func (r *Registry) ComponentFor(componentInstance any) (*RegistrableComponent, b
 	return r.componentFor(componentInstance)
 }
 
-func (r *Registry) ComponentByID(id uint32) (*RegistrableComponent, bool) {
-	return r.componentByID(id)
-}
-
 func (r *Registry) TaskFor(taskInstance any) (*RegistrableTask, bool) {
 	return r.taskFor(taskInstance)
 }

chasm/export_test.go

@@ -6,6 +6,7 @@ import (
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/tests/gen/testspb/v1"
 	"go.temporal.io/server/common"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -157,11 +158,6 @@ func (s *PayloadStore) SearchAttributes(
 }
 
 // Memo implements chasm.VisibilityMemoProvider interface
-func (s *PayloadStore) Memo(
-	_ chasm.Context,
-) map[string]chasm.VisibilityValue {
-	return map[string]chasm.VisibilityValue{
-		TotalCountMemoFieldName: chasm.VisibilityValueInt64(s.State.TotalCount),
-		TotalSizeMemoFieldName:  chasm.VisibilityValueInt64(s.State.TotalSize),
-	}
+func (s *PayloadStore) Memo(_ chasm.Context) proto.Message {
+	return s.State
 }

chasm/lib/tests/payload.go

@@ -85,6 +85,14 @@ func (r *ComponentRef) ArchetypeID(
 	return r.archetypeID, nil
 }
 
+func GetArchetypeIDFromType(registry *Registry, archetypeType reflect.Type) (ArchetypeID, error) {
+	rc, ok := registry.ComponentOf(archetypeType)
+	if !ok {
+		return 0, serviceerror.NewInternal("unknown chasm component type: " + archetypeType.String())
+	}
+	return rc.componentID, nil
+}
+
 // ShardingKey returns the sharding key used for determining the shardID of the run
 // that contains the referenced component.
 // TODO: remove this method and ShardingKey concept, we don't need this functionality.

chasm/ref.go

@@ -81,6 +81,15 @@ func (r *Registry) ComponentFqnByID(id uint32) (string, bool) {
 	return fqn, ok
 }
 
+// ComponentByID returns the registrable component for a given archetype ID.
+func (r *Registry) ComponentByID(id uint32) (*RegistrableComponent, bool) {
+	fqn, ok := r.componentFqnByID[id]
+	if !ok {
+		return nil, false
+	}
+	return r.component(fqn)
+}
+
 // ComponentIDFor converts registered component instance to component type ID.
 // This method should only be used by CHASM framework internal code,
 // NOT CHASM library developers.

chasm/registry.go

@@ -10,6 +10,7 @@ import (
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"google.golang.org/protobuf/proto"
 )
 
 type (
@@ -113,10 +114,8 @@ func (tc *TestComponent) SearchAttributes(_ Context) []SearchAttributeKeyValue {
 }
 
 // Memo implements VisibilityMemoProvider interface.
-func (tc *TestComponent) Memo(_ Context) map[string]VisibilityValue {
-	return map[string]VisibilityValue{
-		TestComponentStartTimeMemoKey: VisibilityValueTime(tc.ComponentData.GetStartTime().AsTime()),
-	}
+func (tc *TestComponent) Memo(_ Context) proto.Message {
+	return tc.ComponentData
 }
 
 func (tsc1 *TestSubComponent1) LifecycleState(_ Context) LifecycleState {

chasm/test_component_test.go

@@ -144,7 +144,7 @@ type (
 		// Root component's search attributes and memo at the start of a transaction.
 		// They will be updated upon CloseTransaction() if they are changed.
 		currentSA   map[string]VisibilityValue
-		currentMemo map[string]VisibilityValue
+		currentMemo proto.Message
 
 		needsPointerResolution bool
 	}
@@ -1473,7 +1473,7 @@ func (n *Node) closeTransactionForceUpdateVisibility(
 	memoProvider, ok := rootComponent.(VisibilityMemoProvider)
 	if ok {
 		newMemo := memoProvider.Memo(immutableContext)
-		if !maps.EqualFunc(n.currentMemo, newMemo, isVisibilityValueEqual) {
+		if !proto.Equal(n.currentMemo, newMemo) {
 			needUpdate = true
 		}
 		n.currentMemo = newMemo

chasm/tree.go

@@ -10,6 +10,7 @@ import (
 	"go.temporal.io/api/serviceerror"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common/payload"
+	"google.golang.org/protobuf/proto"
 )
 
 const (
@@ -35,7 +36,7 @@ type VisibilitySearchAttributesProvider interface {
 // a transaction, if a visibility task needs to be generated to update the
 // visibility record with the returned memo.
 type VisibilityMemoProvider interface {
-	Memo(Context) map[string]VisibilityValue
+	Memo(Context) proto.Message
 }
 
 // VisibilitySearchAttributesMapper is a mapper for CHASM search attributes.

chasm/visibility.go

@@ -11,6 +11,7 @@ import (
 	enumspb "go.temporal.io/api/enums/v1"
 	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/common/debug"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log/tag"
@@ -23,6 +24,7 @@ import (
 	"go.temporal.io/server/common/persistence/visibility/manager"
 	"go.temporal.io/server/common/primitives/timestamp"
 	"go.temporal.io/server/common/resolver"
+	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/common/searchattribute/defs"
 	"go.uber.org/mock/gomock"
 )
@@ -69,6 +71,7 @@ func (s *VisibilityPersistenceSuite) SetupSuite() {
 		s.SearchAttributesProvider,
 		s.SearchAttributesMapperProvider,
 		s.NamespaceRegistry,
+		chasm.NewRegistry(nil),
 		dynamicconfig.GetIntPropertyFn(1000),
 		dynamicconfig.GetIntPropertyFn(1000),
 		dynamicconfig.GetFloatPropertyFn(0.2),
@@ -820,7 +823,7 @@ func (s *VisibilityPersistenceSuite) TestUpsertWorkflowExecution() {
 					Memo:             nil,
 					SearchAttributes: &commonpb.SearchAttributes{
 						IndexedFields: map[string]*commonpb.Payload{
-							searchattribute.TemporalChangeVersion: temporalChangeVersionPayload,
+							defs.TemporalChangeVersion: temporalChangeVersionPayload,
 						},
 					},
 					Status: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,