CHASM: Add ParentPtr field (#8672)

## What changed?
- Add CHASM ParentPtr 

## Why?
- Current implementation of CHASM pointer is a separate physical node.
But in most of the use case we have, we only need to point to the parent
component which can be done entirely in memory without persisting
anything.

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [x] added new unit test(s)
- [ ] added new functional test(s)

Author: yycptt

chasm/fields_iterator.go

@@ -9,9 +9,10 @@ import (
 )
 
 const (
-	chasmFieldTypePrefix = "chasm.Field["
-	chasmMapTypePrefix   = "chasm.Map["
-	chasmMSPointerType   = "chasm.MSPointer"
+	chasmFieldTypePrefix         = "chasm.Field["
+	chasmMapTypePrefix           = "chasm.Map["
+	chasmMSPointerType           = "chasm.MSPointer"
+	chasmParentPointerTypePrefix = "chasm.ParentPtr["
 
 	fieldNameTag = "name"
 )
@@ -24,6 +25,7 @@ const (
 	fieldKindSubField
 	fieldKindSubMap
 	fieldKindMutableState
+	fieldKindParentPtr
 )
 
 type fieldInfo struct {
@@ -62,7 +64,10 @@ func fieldsOf(valueV reflect.Value) iter.Seq[fieldInfo] {
 				prefix := genericTypePrefix(fieldT)
 				if strings.HasPrefix(prefix, "*") {
 					switch prefix[1:] {
-					case chasmFieldTypePrefix, chasmMapTypePrefix, chasmMSPointerType:
+					case chasmFieldTypePrefix,
+						chasmMapTypePrefix,
+						chasmMSPointerType,
+						chasmParentPointerTypePrefix:
 						fieldErr = serviceerror.NewInternalf("%s.%s: CHASM fields must not be pointers", valueT, fieldN)
 					default:
 						continue
@@ -75,6 +80,8 @@ func fieldsOf(valueV reflect.Value) iter.Seq[fieldInfo] {
 						fieldK = fieldKindSubMap
 					case chasmMSPointerType:
 						fieldK = fieldKindMutableState
+					case chasmParentPointerTypePrefix:
+						fieldK = fieldKindParentPtr
 					default:
 						continue // Skip non-CHASM fields.
 					}
@@ -113,7 +120,10 @@ func unmanagedFieldsOf(valueT reflect.Type) iter.Seq[fieldInfo] {
 			fieldN := fieldName(valueT.Field(i))
 			prefix := genericTypePrefix(fieldT)
 			switch prefix {
-			case chasmFieldTypePrefix, chasmMapTypePrefix, chasmMSPointerType:
+			case chasmFieldTypePrefix,
+				chasmMapTypePrefix,
+				chasmMSPointerType,
+				chasmParentPointerTypePrefix:
 				continue // Skip CHASM fields.
 			default:
 				if !yield(fieldInfo{typ: fieldT, name: fieldN}) {

chasm/parent_pointer.go

@@ -0,0 +1,102 @@
+package chasm
+
+import (
+	"fmt"
+	"reflect"
+
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/common/softassert"
+)
+
+const (
+	parentPtrInternalFieldName = "Internal"
+)
+
+// ParentPtr is a in-memory pointer to the parent component of a CHASM component.
+//
+// CHASM map is not a component, so if a component is inside a map, its ParentPtr
+// will point to the nearest ancestor component that is not a map.
+//
+// ParentPtr is only initialized and available for use **after** the transition that
+// creates the component using ParentPtr is completed.
+type ParentPtr[T Component] struct {
+	// Exporting this field as this generic struct needs to be created via reflection,
+	// and reflection can't set private fields.
+	Internal parentPtrInternal
+}
+
+type parentPtrInternal struct {
+	// Storing currentNode instead of parent component Node here so that
+	// we can differentiate between root node and non-initialized ParentPtr.
+	currentNode *Node
+}
+
+// Get returns the parent component, deserializing it if necessary.
+// Panics rather than returning an error, as errors are supposed to be handled by the framework as opposed to the
+// application.
+func (p ParentPtr[T]) Get(chasmContext Context) T {
+	vT, ok := p.TryGet(chasmContext)
+	if !ok {
+		// nolint:forbidigo // Panic is intended here for framework error handling.
+		panic(serviceerror.NewInternal("expect parent component value but got nil"))
+	}
+	return vT
+}
+
+// TryGet returns the parent component and a boolean indicating if the value was found,
+// deserializing if necessary.
+// Panics rather than returning an error, as errors are supposed to be handled by the framework as opposed to the
+// application.
+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"))
+	}
+
+	parent := p.Internal.currentNode.parent
+	if parent == nil {
+		return nilT, false
+	}
+
+	for parent.isMap() {
+		parent = parent.parent
+		if parent == nil {
+			encodedPath, _ := p.Internal.currentNode.getEncodedPath()
+			// nolint:forbidigo // Panic is intended here for framework error handling.
+			panic(softassert.UnexpectedInternalErr(
+				p.Internal.currentNode.logger,
+				"unable to find parent component for CHASM component inside a map",
+				fmt.Errorf("child node name: %s", encodedPath),
+			))
+		}
+	}
+
+	if !parent.isComponent() {
+		// nolint:forbidigo // Panic is intended here for framework error handling.
+		panic(softassert.UnexpectedInternalErr(
+			parent.logger,
+			"unexpected CHASM node that has a child component",
+			fmt.Errorf("node %s, node metadata: %s",
+				parent.nodeName,
+				parent.serializedNode.GetMetadata().String(),
+			),
+		))
+	}
+
+	if err := parent.prepareComponentValue(chasmContext); err != nil {
+		// nolint:forbidigo // Panic is intended here for framework error handling.
+		panic(err)
+	}
+
+	if parent.value == nil {
+		return nilT, false
+	}
+
+	vT, isT := parent.value.(T)
+	if !isT {
+		// nolint:forbidigo // Panic is intended here for framework error handling.
+		panic(serviceerror.NewInternalf("parent component value doesn't implement %s", reflect.TypeFor[T]().Name()))
+	}
+	return vT, true
+}

chasm/test_component_test.go

@@ -31,6 +31,7 @@ type (
 		SubComponentInterfacePointer Field[Component]
 
 		MSPointer MSPointer
+		ParentPtr ParentPtr[*TestComponent]
 
 		Visibility Field[*Visibility]
 	}
@@ -44,12 +45,16 @@ type (
 		SubData11            Field[*protoMessageType] // Random proto message.
 		SubComponent2Pointer Field[*TestSubComponent2]
 		DataPointer          Field[*protoMessageType]
+
+		ParentPtr ParentPtr[*TestComponent]
 	}
 
 	TestSubComponent11 struct {
 		UnimplementedComponent
 
 		SubComponent11Data *protoMessageType
+
+		ParentPtr ParentPtr[*TestSubComponent1]
 	}
 
 	TestSubComponent2 struct {

chasm/tree.go

@@ -538,6 +538,10 @@ func (n *Node) isComponent() bool {
 	return n.serializedNode.GetMetadata().GetComponentAttributes() != nil
 }
 
+func (n *Node) isMap() bool {
+	return n.serializedNode.GetMetadata().GetCollectionAttributes() != nil
+}
+
 func (n *Node) fieldType() fieldType {
 	if n.serializedNode.GetMetadata().GetComponentAttributes() != nil {
 		return fieldTypeComponent
@@ -757,6 +761,19 @@ func (n *Node) syncSubComponents() error {
 			if keepChild {
 				childrenToKeep[field.name] = struct{}{}
 			}
+		case fieldKindParentPtr:
+			internalField := field.val.FieldByName(parentPtrInternalFieldName)
+			internal, ok := internalField.Interface().(parentPtrInternal)
+			if !ok {
+				return softassert.UnexpectedInternalErr(
+					n.logger,
+					"CHASM parent pointer's internal field is not of parentPtrInternal type",
+					fmt.Errorf("node %s, actual type: %T", n.nodeName, internalField.Interface()))
+			}
+			if internal.currentNode == nil || internal.currentNode != n {
+				internal.currentNode = n
+				internalField.Set(reflect.ValueOf(internal))
+			}
 		case fieldKindSubMap:
 			if field.val.IsNil() {
 				// If Map field is nil then delete all collection items nodes and collection node itself.
@@ -1141,6 +1158,12 @@ func (n *Node) deserializeComponentNode(
 			}
 		case fieldKindMutableState:
 			field.val.Set(reflect.ValueOf(NewMSPointer(n.backend)))
+		case fieldKindParentPtr:
+			parentPtrV := reflect.New(field.typ).Elem()
+			parentPtrV.FieldByName(parentPtrInternalFieldName).Set(reflect.ValueOf(parentPtrInternal{
+				currentNode: n,
+			}))
+			field.val.Set(parentPtrV)
 		}
 	}
 

chasm/tree_test.go

@@ -513,6 +513,68 @@ func (s *nodeSuite) TestPointerAttributes() {
 	})
 }
 
+func (s *nodeSuite) TestParentPointer_InMemory() {
+	node := s.testComponentTree()
+
+	s.assertParentPointer(node)
+
+	// Additionally also test parentPtr for components inside a map.
+
+	mutableContext := NewMutableContext(context.Background(), node)
+	component, err := node.Component(mutableContext, ComponentRef{})
+	s.NoError(err)
+	testComponent := component.(*TestComponent)
+
+	mapSubComponent1 := &TestSubComponent1{}
+	// Try using the testComponent we get from the ParentPtr for the mutation.
+	testComponent.SubComponents = Map[string, *TestSubComponent1]{
+		"mapSubComponent1": NewComponentField(mutableContext, mapSubComponent1),
+	}
+
+	s.Panics(func() {
+		_ = mapSubComponent1.ParentPtr.Get(mutableContext)
+	})
+
+	// Sync structure initializes the parent pointer
+	err = node.syncSubComponents()
+	s.NoError(err)
+
+	testComponentFromPtr := mapSubComponent1.ParentPtr.Get(mutableContext)
+	// Asserting they actually point to the same testComponent object.
+	s.Same(testComponent, testComponentFromPtr)
+}
+
+func (s *nodeSuite) TestParentPointer_FromDB() {
+	serializedNodes := testComponentSerializedNodes()
+
+	node, err := s.newTestTree(serializedNodes)
+	s.NoError(err)
+
+	s.assertParentPointer(node)
+}
+
+func (s *nodeSuite) assertParentPointer(testComponentNode *Node) {
+	chasmContext := NewContext(context.Background(), testComponentNode)
+	component, err := testComponentNode.Component(chasmContext, ComponentRef{})
+	s.NoError(err)
+	testComponent := component.(*TestComponent)
+
+	_, found := testComponent.ParentPtr.TryGet(chasmContext)
+	s.False(found)
+
+	subComponent1, err := testComponent.SubComponent1.Get(chasmContext)
+	s.NoError(err)
+	testComponentFromPtr := subComponent1.ParentPtr.Get(chasmContext)
+	// Asserting they actually point to the same testComponent object.
+	s.Same(testComponent, testComponentFromPtr)
+
+	subComponent11, err := subComponent1.SubComponent11.Get(chasmContext)
+	s.NoError(err)
+	testSubComponent1FromPtr := subComponent11.ParentPtr.Get(chasmContext)
+	// Asserting they actually point to the same testSubComponent1 object.
+	s.Same(subComponent1, testSubComponent1FromPtr)
+}
+
 func (s *nodeSuite) TestSyncSubComponents_DeleteLeafNode() {
 	node := s.testComponentTree()