From 7e73933cf0e1675040401803ababb9691a41fe53 Mon Sep 17 00:00:00 2001 From: Roey Berman Date: Thu, 15 Jan 2026 15:39:28 -0800 Subject: [PATCH] Various SAA fixes (#9047) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What changed? - Replaced custom search attribute aliases with standard ones: - ActivityStatus → ExecutionStatus (using the standard execution status field) - ActivityTaskQueue → TaskQueue (using the standard task queue field) - ActivityType now uses the alias name directly instead of a constant - Renamed the GetValue() function to SearchAttributeValue() for better clarity - Changed Activity.Store field type from chasm.Field[ActivityStore] to chasm.ParentPtr[ActivityStore] to avoid the storage overhead of a pointer that is only ever expected to be a parent - Fixed error types in handler.go: changed from serviceerror.NewFailedPrecondition to serviceerror.NewInvalidArgumentf for invalid policy errors - Improved ParentPtr.TryGet() to gracefully return false instead of panicking when not initialized --- chasm/lib/activity/activity.go | 14 ++++---------- chasm/lib/activity/frontend.go | 6 +++--- chasm/lib/activity/handler.go | 5 ++--- chasm/lib/activity/library.go | 2 +- chasm/parent_pointer.go | 4 ++-- chasm/search_attribute.go | 4 ++-- chasm/search_attribute_test.go | 16 ++++++++-------- .../visibility/chasm_visibility_manager_test.go | 2 +- tests/chasm_test.go | 8 ++++---- tests/standalone_activity_test.go | 16 ++++++++-------- 10 files changed, 35 insertions(+), 42 deletions(-) diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index cba5d05cb2..45ccf3ee2a 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -31,16 +31,11 @@ const ( // WorkflowTypeTag is a required workflow tag for standalone activities to ensure consistent // metric labeling between workflows and activities. WorkflowTypeTag = "__temporal_standalone_activity__" - - TypeSAAlias = "ActivityType" - StatusSAAlias = "ActivityStatus" - TaskQueueSAAlias = "ActivityTaskQueue" ) var ( - TypeSearchAttribute = chasm.NewSearchAttributeKeyword(TypeSAAlias, chasm.SearchAttributeFieldKeyword01) - StatusSearchAttribute = chasm.NewSearchAttributeKeyword(StatusSAAlias, chasm.SearchAttributeFieldLowCardinalityKeyword01) - TaskQueueSearchAttribute = chasm.NewSearchAttributeKeyword(TaskQueueSAAlias, chasm.SearchAttributeFieldKeyword02) + TypeSearchAttribute = chasm.NewSearchAttributeKeyword("ActivityType", chasm.SearchAttributeFieldKeyword01) + StatusSearchAttribute = chasm.NewSearchAttributeKeyword("ExecutionStatus", chasm.SearchAttributeFieldLowCardinalityKeyword01) ) var _ chasm.VisibilitySearchAttributesProvider = (*Activity)(nil) @@ -69,9 +64,8 @@ type Activity struct { // Pointer to an implementation of the "store". For a workflow activity this would be a parent // pointer back to the workflow. For a standalone activity this is nil (Activity itself // implements the ActivityStore interface). - // TODO(saa-preview): revisit a standalone activity pointing to itself once we handle storing it more efficiently. // TODO(saa-preview): figure out better naming. - Store chasm.Field[ActivityStore] + Store chasm.ParentPtr[ActivityStore] } // WithToken wraps a request with its deserialized task token. @@ -920,6 +914,6 @@ func (a *Activity) SearchAttributes(_ chasm.Context) []chasm.SearchAttributeKeyV return []chasm.SearchAttributeKeyValue{ TypeSearchAttribute.Value(a.GetActivityType().GetName()), StatusSearchAttribute.Value(InternalStatusToAPIStatus(a.GetStatus()).String()), - TaskQueueSearchAttribute.Value(a.GetTaskQueue().GetName()), + chasm.SearchAttributeTaskQueue.Value(a.GetTaskQueue().GetName()), } } diff --git a/chasm/lib/activity/frontend.go b/chasm/lib/activity/frontend.go index 120719492a..076f13492a 100644 --- a/chasm/lib/activity/frontend.go +++ b/chasm/lib/activity/frontend.go @@ -193,9 +193,9 @@ func (h *frontendHandler) ListActivityExecutions( executions := make([]*apiactivitypb.ActivityExecutionListInfo, 0, len(resp.Executions)) for _, exec := range resp.Executions { - activityType, _ := chasm.GetValue(exec.ChasmSearchAttributes, TypeSearchAttribute) - taskQueue, _ := chasm.GetValue(exec.ChasmSearchAttributes, TaskQueueSearchAttribute) - statusStr, _ := chasm.GetValue(exec.ChasmSearchAttributes, StatusSearchAttribute) + activityType, _ := chasm.SearchAttributeValue(exec.ChasmSearchAttributes, TypeSearchAttribute) + taskQueue, _ := chasm.SearchAttributeValue(exec.ChasmSearchAttributes, chasm.SearchAttributeTaskQueue) + statusStr, _ := chasm.SearchAttributeValue(exec.ChasmSearchAttributes, StatusSearchAttribute) status, _ := enumspb.ActivityExecutionStatusFromString(statusStr) info := &apiactivitypb.ActivityExecutionListInfo{ diff --git a/chasm/lib/activity/handler.go b/chasm/lib/activity/handler.go index 2806599701..be594126bb 100644 --- a/chasm/lib/activity/handler.go +++ b/chasm/lib/activity/handler.go @@ -3,7 +3,6 @@ package activity import ( "context" "errors" - "fmt" enumspb "go.temporal.io/api/enums/v1" errordetailspb "go.temporal.io/api/errordetails/v1" @@ -58,12 +57,12 @@ func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.St reusePolicy, ok := businessIDReusePolicyMap[frontendReq.GetIdReusePolicy()] if !ok { - return nil, serviceerror.NewFailedPrecondition(fmt.Sprintf("unsupported ID reuse policy: %v", frontendReq.GetIdReusePolicy())) + return nil, serviceerror.NewInvalidArgumentf("unsupported ID reuse policy: %v", frontendReq.GetIdReusePolicy()) } conflictPolicy, ok := businessIDConflictPolicyMap[frontendReq.GetIdConflictPolicy()] if !ok { - return nil, serviceerror.NewFailedPrecondition(fmt.Sprintf("unsupported ID conflict policy: %v", frontendReq.GetIdConflictPolicy())) + return nil, serviceerror.NewInvalidArgumentf("unsupported ID conflict policy: %v", frontendReq.GetIdConflictPolicy()) } result, err := chasm.NewExecution( diff --git a/chasm/lib/activity/library.go b/chasm/lib/activity/library.go index 5c5bf2bb5f..6fa017d1b2 100644 --- a/chasm/lib/activity/library.go +++ b/chasm/lib/activity/library.go @@ -34,7 +34,7 @@ func (l *componentOnlyLibrary) Components() []*chasm.RegistrableComponent { chasm.WithSearchAttributes( TypeSearchAttribute, StatusSearchAttribute, - TaskQueueSearchAttribute, + chasm.SearchAttributeTaskQueue, ), chasm.WithBusinessIDAlias("ActivityId"), ), diff --git a/chasm/parent_pointer.go b/chasm/parent_pointer.go index 7c7cdb2c93..b2717a0d8c 100644 --- a/chasm/parent_pointer.go +++ b/chasm/parent_pointer.go @@ -50,8 +50,8 @@ func (p ParentPtr[T]) Get(chasmContext Context) T { func (p ParentPtr[T]) TryGet(chasmContext Context) (T, bool) { var nilT T if p.Internal.currentNode == nil { - // nolint:forbidigo // Panic is intended here for framework error handling. - panic(serviceerror.NewInternal("parent pointer not initialized yet")) + // ParentPtr not initialized + return nilT, false } parent := p.Internal.currentNode.parent diff --git a/chasm/search_attribute.go b/chasm/search_attribute.go index 273bb92745..b8c6238720 100644 --- a/chasm/search_attribute.go +++ b/chasm/search_attribute.go @@ -422,11 +422,11 @@ func NewSearchAttributesMap(values map[string]VisibilityValue) SearchAttributesM return SearchAttributesMap{values: values} } -// GetValue returns the value for a given SearchAttribute with compile-time type safety. +// SearchAttributeValue returns the value for a given SearchAttribute with compile-time type safety. // The return type T is inferred from the SearchAttribute's type parameter. // For example, SearchAttributeBool will return a bool value. // If the value is not found or the type does not match, the zero value for the type T is returned and the second return value is false. -func GetValue[T any](m SearchAttributesMap, sa typedSearchAttribute[T]) (val T, ok bool) { +func SearchAttributeValue[T any](m SearchAttributesMap, sa typedSearchAttribute[T]) (val T, ok bool) { var zero T if len(m.values) == 0 { return zero, false diff --git a/chasm/search_attribute_test.go b/chasm/search_attribute_test.go index d0ada33be2..0f539135e7 100644 --- a/chasm/search_attribute_test.go +++ b/chasm/search_attribute_test.go @@ -30,51 +30,51 @@ func TestSearchAttributesMap_Get(t *testing.T) { m := NewSearchAttributesMap(values) t.Run("GetBool", func(t *testing.T) { - val, ok := GetValue(m, boolAttr) + val, ok := SearchAttributeValue(m, boolAttr) assert.True(t, ok) assert.True(t, val) }) t.Run("GetInt64", func(t *testing.T) { - val, ok := GetValue(m, intAttr) + val, ok := SearchAttributeValue(m, intAttr) assert.True(t, ok) assert.Equal(t, int64(42), val) }) t.Run("GetFloat64", func(t *testing.T) { - val, ok := GetValue(m, doubleAttr) + val, ok := SearchAttributeValue(m, doubleAttr) assert.True(t, ok) assert.InDelta(t, 3.14, val, 0.0001) }) t.Run("GetString", func(t *testing.T) { - val, ok := GetValue(m, keywordAttr) + val, ok := SearchAttributeValue(m, keywordAttr) assert.True(t, ok) assert.Equal(t, "active", val) }) t.Run("GetTime", func(t *testing.T) { - val, ok := GetValue(m, datetimeAttr) + val, ok := SearchAttributeValue(m, datetimeAttr) assert.True(t, ok) assert.True(t, now.Equal(val)) }) t.Run("GetStringSlice", func(t *testing.T) { - val, ok := GetValue(m, keywordListAttr) + val, ok := SearchAttributeValue(m, keywordListAttr) assert.True(t, ok) assert.Equal(t, []string{"tag1", "tag2"}, val) }) t.Run("NotFound", func(t *testing.T) { missingAttr := NewSearchAttributeBool("missing", SearchAttributeFieldBool02) - val, ok := GetValue(m, missingAttr) + val, ok := SearchAttributeValue(m, missingAttr) assert.False(t, ok) assert.False(t, val) }) t.Run("NilMap", func(t *testing.T) { emptyMap := NewSearchAttributesMap(nil) - val, ok := GetValue(emptyMap, boolAttr) + val, ok := SearchAttributeValue(emptyMap, boolAttr) assert.False(t, ok) assert.False(t, val) }) diff --git a/common/persistence/visibility/chasm_visibility_manager_test.go b/common/persistence/visibility/chasm_visibility_manager_test.go index 7fc0db43cd..cbbdf8b44d 100644 --- a/common/persistence/visibility/chasm_visibility_manager_test.go +++ b/common/persistence/visibility/chasm_visibility_manager_test.go @@ -482,7 +482,7 @@ func (s *ChasmVisibilityManagerSuite) TestListExecutions_WithTaskQueueSearchAttr execution := response.Executions[0] s.NotNil(execution.ChasmSearchAttributes) // Verify TaskQueue is in the CHASM search attributes using GetValue with the preallocated search attribute - taskQueueVal, ok := chasm.GetValue(execution.ChasmSearchAttributes, chasm.SearchAttributeTaskQueue) + taskQueueVal, ok := chasm.SearchAttributeValue(execution.ChasmSearchAttributes, chasm.SearchAttributeTaskQueue) s.True(ok) s.Equal(expectedTaskQueue, taskQueueVal) } diff --git a/tests/chasm_test.go b/tests/chasm_test.go index 7830242743..dac2ff3acb 100644 --- a/tests/chasm_test.go +++ b/tests/chasm_test.go @@ -259,10 +259,10 @@ func (s *ChasmTestSuite) TestListExecutions() { s.Equal(0, int(totalCount)) totalSize := visRecord.ChasmMemo.TotalSize s.Equal(0, int(totalSize)) - totalCountSA, ok := chasm.GetValue(visRecord.ChasmSearchAttributes, tests.PayloadTotalCountSearchAttribute) + totalCountSA, ok := chasm.SearchAttributeValue(visRecord.ChasmSearchAttributes, tests.PayloadTotalCountSearchAttribute) s.True(ok) s.Equal(0, int(totalCountSA)) - totalSizeSA, ok := chasm.GetValue(visRecord.ChasmSearchAttributes, tests.PayloadTotalSizeSearchAttribute) + totalSizeSA, ok := chasm.SearchAttributeValue(visRecord.ChasmSearchAttributes, tests.PayloadTotalSizeSearchAttribute) s.True(ok) s.Equal(0, int(totalSizeSA)) var scheduledByID string @@ -636,7 +636,7 @@ func (s *ChasmTestSuite) TestListExecutions_ExecutionStatusAsAlias() { s.Equal(storeID, visRecord.BusinessID) // Verify the ExecutionStatus CHASM search attribute is correctly returned - executionStatus, ok := chasm.GetValue(visRecord.ChasmSearchAttributes, tests.ExecutionStatusSearchAttribute) + executionStatus, ok := chasm.SearchAttributeValue(visRecord.ChasmSearchAttributes, tests.ExecutionStatusSearchAttribute) s.True(ok) s.Equal("Running", executionStatus) @@ -714,7 +714,7 @@ func (s *ChasmTestSuite) TestTaskQueuePreallocatedSearchAttribute() { s.Equal(storeID, visRecord.BusinessID) // Verify TaskQueue is returned as a CHASM search attribute - taskQueueVal, ok := chasm.GetValue(visRecord.ChasmSearchAttributes, chasm.SearchAttributeTaskQueue) + taskQueueVal, ok := chasm.SearchAttributeValue(visRecord.ChasmSearchAttributes, chasm.SearchAttributeTaskQueue) s.True(ok) s.Equal(tests.DefaultPayloadStoreTaskQueue, taskQueueVal) } diff --git a/tests/standalone_activity_test.go b/tests/standalone_activity_test.go index b245cbc0e7..1b6e06536e 100644 --- a/tests/standalone_activity_test.go +++ b/tests/standalone_activity_test.go @@ -2399,12 +2399,12 @@ func (s *standaloneActivityTestSuite) TestListActivityExecutions() { verifyListQuery(t, fmt.Sprintf("ActivityType = '%s'", activityType), 10) }) - t.Run("QueryByActivityStatus", func(t *testing.T) { - verifyListQuery(t, fmt.Sprintf("ActivityStatus = 'Running' AND ActivityType = '%s'", activityType), 10) + t.Run("QueryByExecutionStatus", func(t *testing.T) { + verifyListQuery(t, fmt.Sprintf("ExecutionStatus = 'Running' AND ActivityType = '%s'", activityType), 10) }) t.Run("QueryByTaskQueue", func(t *testing.T) { - verifyListQuery(t, fmt.Sprintf("ActivityTaskQueue = '%s' AND ActivityType = '%s'", taskQueue, activityType), 10) + verifyListQuery(t, fmt.Sprintf("TaskQueue = '%s' AND ActivityType = '%s'", taskQueue, activityType), 10) }) t.Run("QueryByMultipleFields", func(t *testing.T) { @@ -2585,15 +2585,15 @@ func (s *standaloneActivityTestSuite) TestCountActivityExecutions() { verifyCountQuery(t, fmt.Sprintf("ActivityType = '%s'", activityType), 1) }) - t.Run("CountByActivityStatus", func(t *testing.T) { - verifyCountQuery(t, fmt.Sprintf("ActivityStatus = 'Running' AND ActivityType = '%s'", activityType), 1) + t.Run("CountByExecutionStatus", func(t *testing.T) { + verifyCountQuery(t, fmt.Sprintf("ExecutionStatus = 'Running' AND ActivityType = '%s'", activityType), 1) }) t.Run("CountByTaskQueue", func(t *testing.T) { - verifyCountQuery(t, fmt.Sprintf("ActivityTaskQueue = '%s' AND ActivityType = '%s'", s.tv.TaskQueue().GetName(), activityType), 1) + verifyCountQuery(t, fmt.Sprintf("TaskQueue = '%s' AND ActivityType = '%s'", s.tv.TaskQueue().GetName(), activityType), 1) }) - t.Run("GroupByActivityStatus", func(t *testing.T) { + t.Run("GroupByExecutionStatus", func(t *testing.T) { groupByType := &commonpb.ActivityType{Name: "count-groupby-test-type"} taskQueue := s.tv.TaskQueue().GetName() @@ -2604,7 +2604,7 @@ func (s *standaloneActivityTestSuite) TestCountActivityExecutions() { require.NotEmpty(t, resp.GetRunId()) } - query := fmt.Sprintf("ActivityType = '%s' GROUP BY ActivityStatus", groupByType.Name) + query := fmt.Sprintf("ActivityType = '%s' GROUP BY ExecutionStatus", groupByType.Name) var resp *workflowservice.CountActivityExecutionsResponse s.Eventually( func() bool {