[CHASM] Pure task processing - GetPureTasks, ExecutePureTasks

chasm/tree.go

@@ -1241,6 +1241,83 @@
 	return false
 }
 
+// GetPureTasks returns all valid, expired/runnable pure tasks within the CHASM
+// tree. The CHASM tree is left untouched, even if invalid tasks are
+// detected (these are cleaned up as part of transaction close).
+func (n *Node) GetPureTasks(deadline time.Time) ([]any, error) {
+	var componentTasks []*persistencespb.ChasmComponentAttributes_Task
+
+	// Walk the tree to find runnable, valid tasks.
+	err := n.walk(func(node *Node) error {
+		// Skip nodes that aren't serialized yet.
+		if node.serializedNode == nil || node.serializedNode.Metadata == nil {
+			return nil
+		}
+
+		componentAttr := node.serializedNode.Metadata.GetComponentAttributes()
+		// Skip nodes that aren't components.
+		if componentAttr == nil {
+			return nil
+		}
+
+		validateContext := NewContext(context.Background(), n)
+		for _, task := range componentAttr.GetPureTasks() {
+			if task.ScheduledTime.AsTime().After(deadline) {
+				// Pure tasks are stored in-order, so we can skip scanning the rest once we hit
+				// an unexpired task deadline.
+				break
+			}
+
+			if task.PhysicalTaskStatus != physicalTaskStatusCreated {
+				continue
+			}
+
+			// Component value must be prepared for validation to work.
+			if err := node.prepareComponentValue(validateContext); err != nil {
+				return err
+			}
+
+			// Validate the task. If the task is invalid, skip it for processing (it'll be
+			// removed when the transaction closes).
+			ok, err := node.validateComponentTask(validateContext, task)
+			if !ok {
+				continue
+			}
+			if err != nil {
+				return err
+			}
+
+			componentTasks = append(componentTasks, task)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// Map serialized component tasks to their deserialized values using the Registry.
+	taskValues := make([]any, len(componentTasks))
+	for idx, componentTask := range componentTasks {
+		registrableTask, ok := n.registry.task(componentTask.GetType())
+		if !ok {
+			return nil, serviceerror.NewInternal(fmt.Sprintf(
+				"unregistered CHASM task type '%s'",
+				componentTask.GetType(),
+			))
+		}
+
+		// TODO - validateComponentTask also calls deserializeTask, should share a cached value
+		taskValue, err := deserializeTask(registrableTask, componentTask.Data)
+		if err != nil {
+			return nil, err
+		}
+		taskValues[idx] = taskValue.Interface()
+	}
+
+	return taskValues, nil
+}
+
 func newNode(
 	base *nodeBase,
 	parent *Node,
@@ -1433,3 +1510,42 @@
 
 	return blob, nil
 }
+
+// ExecutePureTask executes the given taskInstance against the node's component.
+func (n *Node) ExecutePureTask(taskInstance any) error {
+	registrableTask, ok := n.registry.taskFor(taskInstance)
+	if !ok {
+		return fmt.Errorf("unknown task type for task instance goType '%s'", reflect.TypeOf(taskInstance).Name())
+	}
+
+	if !registrableTask.isPureTask {
+		return fmt.Errorf("ExecutePureTask called on a SideEffect task '%s'", registrableTask.fqType())
+	}
+
+	// Ensure this node's component value is hydrated before execution.
+	ctx := NewContext(context.Background(), n)
+	if err := n.prepareComponentValue(ctx); err != nil {
+		return err
+	}
+
+	// TODO - access rule check here?
+	// TODO - instantiate CHASM engine and attach to context
+
+	executor := registrableTask.handler
+	if executor == nil {
+		return fmt.Errorf("no handler registered for task type '%s'", registrableTask.taskType)
+	}
+
+	fn := reflect.ValueOf(executor).MethodByName("Execute")
+	result := fn.Call([]reflect.Value{
+		reflect.ValueOf(ctx),
+		reflect.ValueOf(n.value),
+		reflect.ValueOf(taskInstance),
+	})
+	if !result[0].IsNil() {
+		//nolint:revive // type cast result is unchecked
+		return result[0].Interface().(error)
+	}
+
+	return nil
+}

chasm/tree_test.go

@@ -1438,3 +1438,157 @@ func (s *nodeSuite) testComponentTree() *Node {
 
 	return node // maybe tc too
 }
+
+func (s *nodeSuite) TestGetPureTasks() {
+	now := s.timeSource.Now()
+
+	payload := &commonpb.Payload{
+		Data: []byte("some-random-data"),
+	}
+	taskBlob, err := serialization.ProtoEncodeBlob(payload, enumspb.ENCODING_TYPE_PROTO3)
+	s.NoError(err)
+
+	// Set up a tree with expired and unexpired pure tasks.
+	persistenceNodes := map[string]*persistencespb.ChasmNode{
+		"": {
+			Metadata: &persistencespb.ChasmNodeMetadata{
+				InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
+				Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
+					ComponentAttributes: &persistencespb.ChasmComponentAttributes{
+						Type: "TestLibrary.test_component",
+						PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
+							{
+								// Expired
+								Type:                      "TestLibrary.test_pure_task",
+								ScheduledTime:             timestamppb.New(now),
+								VersionedTransition:       &persistencespb.VersionedTransition{TransitionCount: 1},
+								VersionedTransitionOffset: 1,
+								PhysicalTaskStatus:        physicalTaskStatusCreated,
+								Data:                      taskBlob,
+							},
+						},
+					},
+				},
+			},
+		},
+		"child": {
+			Metadata: &persistencespb.ChasmNodeMetadata{
+				InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
+				Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
+					ComponentAttributes: &persistencespb.ChasmComponentAttributes{
+						Type: "TestLibrary.test_component",
+						PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
+							{
+								Type: "TestLibrary.test_pure_task",
+								// Unexpired
+								ScheduledTime:             timestamppb.New(now.Add(time.Hour)),
+								VersionedTransition:       &persistencespb.VersionedTransition{TransitionCount: 1},
+								VersionedTransitionOffset: 1,
+								PhysicalTaskStatus:        physicalTaskStatusCreated,
+								Data:                      taskBlob,
+							},
+						},
+					},
+				},
+			},
+		},
+		"child/grandchild1": {
+			Metadata: &persistencespb.ChasmNodeMetadata{
+				InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
+				Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
+					ComponentAttributes: &persistencespb.ChasmComponentAttributes{
+						Type: "TestLibrary.test_component",
+						PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
+							{
+								Type: "TestLibrary.test_pure_task",
+								// Expired, but physical task not created
+								ScheduledTime:             timestamppb.New(now),
+								VersionedTransition:       &persistencespb.VersionedTransition{TransitionCount: 1},
+								VersionedTransitionOffset: 2,
+								PhysicalTaskStatus:        physicalTaskStatusNone,
+								Data:                      taskBlob,
+							},
+							{
+								Type: "TestLibrary.test_pure_task",
+								// Expired
+								ScheduledTime:             timestamppb.New(now),
+								VersionedTransition:       &persistencespb.VersionedTransition{TransitionCount: 1},
+								VersionedTransitionOffset: 1,
+								PhysicalTaskStatus:        physicalTaskStatusCreated,
+								Data:                      taskBlob,
+							},
+							{
+								Type: "TestLibrary.test_pure_task",
+								// Expired, but will fail to validate
+								ScheduledTime:             timestamppb.New(now),
+								VersionedTransition:       &persistencespb.VersionedTransition{TransitionCount: 1},
+								VersionedTransitionOffset: 1,
+								PhysicalTaskStatus:        physicalTaskStatusCreated,
+								Data:                      taskBlob,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	rt, ok := s.registry.Task("TestLibrary.test_pure_task")
+	s.True(ok)
+
+	// Succeed first two task validations, and then fail the third.
+	rt.validator.(*MockTaskValidator[any, *TestPureTask]).EXPECT().
+		Validate(gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(2)
+	rt.validator.(*MockTaskValidator[any, *TestPureTask]).EXPECT().
+		Validate(gomock.Any(), gomock.Any(), gomock.Any()).Return(false, nil).Times(1)
+
+	root, err := NewTree(persistenceNodes, s.registry, s.timeSource, s.nodeBackend, s.nodePathEncoder, s.logger)
+	s.NoError(err)
+	s.NotNil(root)
+
+	tasks, err := root.GetPureTasks(now.Add(time.Minute))
+	s.NoError(err)
+	s.NotNil(tasks)
+	s.Equal(2, len(tasks))
+
+	_, ok = tasks[0].(*TestPureTask)
+	s.True(ok)
+}
+
+func (s *nodeSuite) TestExecutePureTask() {
+	persistenceNodes := map[string]*persistencespb.ChasmNode{
+		"": {
+			Metadata: &persistencespb.ChasmNodeMetadata{
+				InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
+				Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
+					ComponentAttributes: &persistencespb.ChasmComponentAttributes{
+						Type: "TestLibrary.test_component",
+					},
+				},
+			},
+		},
+	}
+
+	pureTask := &TestPureTask{
+		Payload: &commonpb.Payload{
+			Data: []byte("some-random-data"),
+		},
+	}
+
+	rt, ok := s.registry.Task("TestLibrary.test_pure_task")
+	s.True(ok)
+
+	rt.handler.(*MockPureTaskExecutor[any, *TestPureTask]).EXPECT().
+		Execute(
+			gomock.Any(),
+			gomock.AssignableToTypeOf(&TestComponent{}),
+			gomock.Eq(pureTask),
+		).Return(nil).Times(1)
+
+	root, err := NewTree(persistenceNodes, s.registry, s.timeSource, s.nodeBackend, s.nodePathEncoder, s.logger)
+	s.NoError(err)
+	s.NotNil(root)
+
+	err = root.ExecutePureTask(pureTask)
+	s.NoError(err)
+}

service/history/interfaces/chasm_tree.go

@@ -3,8 +3,10 @@
 package interfaces
 
 import (
+	"time"
+
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
 )
 
 var _ ChasmTree = (*chasm.Node)(nil)
@@ -15,4 +17,7 @@
 	ApplyMutation(chasm.NodesMutation) error
 	ApplySnapshot(chasm.NodesSnapshot) error
 	IsDirty() bool
+
+	GetPureTasks(deadline time.Time) ([]any, error)
+	ExecutePureTask(taskInstance any) error
 }

service/history/workflow/noop_chasm_tree.go

@@ -1,6 +1,8 @@
 package workflow
 
 import (
+	"time"
+
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
 	historyi "go.temporal.io/server/service/history/interfaces"
@@ -29,3 +31,11 @@ func (*noopChasmTree) ApplySnapshot(chasm.NodesSnapshot) error {
 func (*noopChasmTree) IsDirty() bool {
 	return false
 }
+
+func (*noopChasmTree) GetPureTasks(deadline time.Time) ([]any, error) {
+	return nil, nil
+}
+
+func (*noopChasmTree) ExecutePureTask(taskInstance any) error {
+	return nil
+}