Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Various SAA fixes (#9047)

Browse Source 
## 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
This commit is contained in:
Roey Berman
2026-01-15 15:39:28 -08:00
committed by GitHub
parent 08f90d3960
commit 7e73933cf0
10 changed files with 35 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
chasm/lib/activity/activity.go
View File

@@ -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()),
}
}

6
chasm/lib/activity/frontend.go
View File

@@ -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{

5
chasm/lib/activity/handler.go
View File

@@ -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(

2
chasm/lib/activity/library.go
View File

@@ -34,7 +34,7 @@ func (l *componentOnlyLibrary) Components() []*chasm.RegistrableComponent {
chasm.WithSearchAttributes(
TypeSearchAttribute,
StatusSearchAttribute,
TaskQueueSearchAttribute,
chasm.SearchAttributeTaskQueue,
),
chasm.WithBusinessIDAlias("ActivityId"),
),

4
chasm/parent_pointer.go
View File

@@ -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

4
chasm/search_attribute.go
View File

@@ -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

16
chasm/search_attribute_test.go
View File

@@ -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)
})

2
common/persistence/visibility/chasm_visibility_manager_test.go
View File

@@ -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)
}

8
tests/chasm_test.go
View File

@@ -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)
}

16
tests/standalone_activity_test.go
View File

@@ -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 {