TestMixedPointerScenario

301 logger log.Logger,

302 metricsHandler metrics.Handler,

303 > ) *Node { tree.go

304 > root := newTreeHelper(registry, timeSource, backend, pathEncoder, logger, metricsHandler)

305 >

306 > // If serializedNodes is empty, it means that this new tree.

307 > // Initialize empty serializedNode.

308 > root.initSerializedNode(fieldTypeComponent)

309 > // Default to Workflow archetype as empty tree is created for workflow as well.

310 > root.serializedNode.Metadata.GetComponentAttributes().TypeId = WorkflowArchetypeID

311 > // Although both value and serializedNode.Data are nil, they are considered NOT synced

312 > // because value has no type and serializedNode does.

313 > // deserialize method should set value when called.

314 > root.setValueState(valueStateNeedDeserialize)

315 > return root

316 > }

317

318 func newTreeHelper(

323 logger log.Logger,

324 metricsHandler metrics.Handler,

325 > ) *Node { tree.go

326 > base := &nodeBase{

327 > registry: registry,

328 > timeSource: timeSource,

329 > backend: backend,

330 > pathEncoder: pathEncoder,

331 > logger: logger,

332 > metricsHandler: metricsHandler,

333 >

334 > mutation: NodesMutation{

335 > UpdatedNodes: make(map[string]*persistencespb.ChasmNode),

336 > DeletedNodes: make(map[string]struct{}),

337 > },

338 > systemMutation: NodesMutation{

339 > UpdatedNodes: make(map[string]*persistencespb.ChasmNode),

340 > DeletedNodes: make(map[string]struct{}),

341 > },

342 > newTasks: make(map[any][]taskWithAttributes),

343 > immediatePureTasks: make(map[any][]taskWithAttributes),

344 > pendingRequestLinks: make(map[any]map[string][]*commonpb.Link),

345 > pendingUserMetadata: make(map[any]*sdkpb.UserMetadata),

346 > valueToNode: make(map[any]*Node),

347 > taskValueCache: make(map[*commonpb.DataBlob]reflect.Value),

348 > needsPointerResolution: false,

349 > }

350 >

351 > return newNode(base, nil, "")

352 > }

353

354 func newTreeInitSearchAttributesAndMemo(

377 }

378

379 > func searchAttributeKeyValuesToMap(saSlice []SearchAttributeKeyValue) map[string]VisibilityValue { tree.go

380 > result := make(map[string]VisibilityValue, len(saSlice))

381 > for _, sa := range saSlice {

382 > result[sa.Field] = sa.Value

383 > }

384 > return result

385 }

386

387 func (n *Node) SetRootComponent(

388 rootComponent RootComponent,

389 > ) error { tree.go

390 > root := n.root()

391 > root.setValue(rootComponent)

392 > root.setValueState(valueStateNeedSyncStructure)

393 > if componentID, ok := n.registry.ComponentIDFor(rootComponent); ok {

394 > root.serializedNode.GetMetadata().GetComponentAttributes().TypeId = componentID

395 > }

396 > return root.syncSubComponents()

397 }

398

400 // If the node is a component or data node, the index from node value to node (valueToNode)

401 // is also updated.

402 > func (n *Node) setValue(value any) { tree.go

403 > if !n.isComponent() && !n.isData() {

404 > n.value = value tree.go

405 > return

406 > }

407

408 > if n.value != nil { tree.go

409 delete(n.valueToNode, n.value)

410 }

411

412 > n.value = value tree.go

413 >

414 > if value != nil {

415 > n.valueToNode[value] = n

416 > }

417 }

418

419 > func (n *Node) setValueState(state valueState) { tree.go

420 > n.valueState = state

421 > if state >= valueStateNeedSerialize {

422 > n.markSubtreeDirty() tree.go

423 > }

424 }

425

429 // markSubtreeDirty marks this node and its entire lineage (ancestors and descendants)

430 // as dirty so that CloseTransaction knows to validate their tasks.

431 > func (n *Node) markSubtreeDirty() { tree.go

432 > // Propagate upward to ancestors.

433 > for cur := n; cur != nil && !cur.subtreeIsDirty; cur = cur.parent {

434 > cur.subtreeIsDirty = true

435 > }

436 // Propagate downward to descendants.

437 > for _, desc := range n.andAllChildren() { tree.go

438 > desc.subtreeIsDirty = true

439 > }

440 }

441

458 chasmContext Context,

459 ref ComponentRef,

460 > ) (Component, error) { tree.go

461 > // Archetype is already validated before this method is called.

462 > // (when the mutable state is loaded, in chasm engine implementation)

463 >

464 > node, ok := n.findNode(ref.componentPath)

465 > if !ok {

466 return nil, errComponentNotFound

467 }

468

469 > if ref.componentInitialVT != nil && transitionhistory.Compare( tree.go

470 > ref.componentInitialVT,

471 > node.serializedNode.Metadata.InitialVersionedTransition,

472 > ) != 0 {

473 return nil, errComponentNotFound

474 }

475

476 > validationContext := NewContext(chasmContext.goContext(), node) tree.go

477 > if err := node.prepareComponentValue(validationContext); err != nil {

478 return nil, err

479 }

480

481 > componentValue, ok := node.value.(Component) tree.go

482 > if !ok {

483 return nil, softassert.UnexpectedInternalErr(

484 n.logger,

487 }

488

489 > if err := node.validateAccess(validationContext, false); err != nil { tree.go

490 return nil, err

491 }

492

493 > if ref.validationFn != nil { tree.go

494 if err := ref.validationFn(node.root().backend, validationContext, componentValue, node.registry); err != nil {

495 return nil, err

498

499 // prepare component value again using incoming context to mark node as dirty if needed.

500 > if err := node.prepareComponentValue(chasmContext); err != nil { tree.go

501 return nil, err

502 }

503 > return componentValue, nil tree.go

504 }

505

517 // Note: engine mutations on paused components are still accepted (checkPaused=false),

518 // per the current requirement.

519 > func (n *Node) validateAccess(ctx Context, checkPaused bool) error { tree.go

520 > intent := operationIntentFromContext(ctx.goContext())

521 > if intent != OperationIntentProgress {

522 > // Read-only operations are always allowed. tree.go

523 > return nil

524 > }

525

526 // Detached nodes skip ancestor validation entirely.

598 func (n *Node) prepareComponentValue(

599 chasmContext Context,

600 > ) error { tree.go

601 > if n.valueState == valueStateNeedDeserialize {

602 metadata := n.serializedNode.Metadata

603 componentAttr := metadata.GetComponentAttributes()

624 // For now, we assume if a node is accessed with a MutableContext,

625 // its value will be mutated and no longer in sync with the serializedNode.

626 > _, componentCanBeMutated := chasmContext.(MutableContext) tree.go

627 > if componentCanBeMutated {

628 > n.setValueState(valueStateNeedSyncStructure) tree.go

629 > }

630

631 > return nil tree.go

632 }

633

661 }

662

663 > func (n *Node) preparePointerValue() error { tree.go

664 > metadata := n.serializedNode.Metadata

665 > pointerAttr := metadata.GetPointerAttributes()

666 > if pointerAttr == nil {

667 return softassert.UnexpectedInternalErr(

668 n.logger,

671 }

672

673 > if n.valueState == valueStateNeedDeserialize { tree.go

674 if err := n.deserialize(nil); err != nil {

675 return fmt.Errorf("failed to deserialize data: %w", err)

677 }

678

679 > return nil tree.go

680 }

681

682 > func (n *Node) isComponent() bool { tree.go

683 > return n.serializedNode.GetMetadata().GetComponentAttributes() != nil

684 > }

685

686 > func (n *Node) isData() bool { tree.go

687 > return n.serializedNode.GetMetadata().GetDataAttributes() != nil

688 > }

689

690 > func (n *Node) isMap() bool { tree.go

691 > return n.serializedNode.GetMetadata().GetCollectionAttributes() != nil

692 > }

693

694 func (n *Node) isDetached() bool {

707 }

708

709 > func (n *Node) fieldType() fieldType { tree.go

710 > if n.serializedNode.GetMetadata().GetComponentAttributes() != nil {

711 > return fieldTypeComponent tree.go

712 > }

713

714 > if n.serializedNode.GetMetadata().GetDataAttributes() != nil { tree.go

715 return fieldTypeData

716 }

717

718 > if n.serializedNode.GetMetadata().GetPointerAttributes() != nil { tree.go

719 > return fieldTypePointer

720 > }

721

722 if n.serializedNode.GetMetadata().GetCollectionAttributes() != nil {

729 }

730

731 > func (n *Node) valueFields() iter.Seq[fieldInfo] { tree.go

732 > return fieldsOf(reflect.ValueOf(n.value))

733 > }

734

735 > func assertStructPointer(t reflect.Type) error { tree.go

736 > if t == nil {

737 return nil

738 }

739

740 > if t.Kind() != reflect.Pointer || t.Elem().Kind() != reflect.Struct { tree.go

741 return serviceerror.NewInternalf("only pointer to struct is supported for tree node value: got %s", t.String())

742 }

743 > return nil tree.go

744 }

745

746 > func (n *Node) initSerializedNode(ft fieldType) { tree.go

747 > switch ft {

748 case fieldTypeData:

749 n.serializedNode = &persistencespb.ChasmNode{

758 },

759 }

760 > case fieldTypeComponent: tree.go

761 > n.serializedNode = &persistencespb.ChasmNode{

762 > Metadata: &persistencespb.ChasmNodeMetadata{

763 > InitialVersionedTransition: &persistencespb.VersionedTransition{

764 > TransitionCount: n.backend.NextTransitionCount(),

765 > NamespaceFailoverVersion: n.backend.GetCurrentVersion(),

766 > },

767 > Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{

768 > ComponentAttributes: &persistencespb.ChasmComponentAttributes{},

769 > },

770 > },

771 > }

772 > case fieldTypePointer, fieldTypeDeferredPointer: tree.go

773 > // A deferred pointer will be resolved to a regular pointer before persistence.

774 > n.serializedNode = &persistencespb.ChasmNode{

775 > Metadata: &persistencespb.ChasmNodeMetadata{

776 > InitialVersionedTransition: &persistencespb.VersionedTransition{

777 > TransitionCount: n.backend.NextTransitionCount(),

778 > NamespaceFailoverVersion: n.backend.GetCurrentVersion(),

779 > },

780 > Attributes: &persistencespb.ChasmNodeMetadata_PointerAttributes{

781 > PointerAttributes: &persistencespb.ChasmPointerAttributes{},

782 > },

783 > },

784 > }

785 case fieldTypeUnspecified:

786 softassert.Fail(n.logger,

827 // effects that must be persisted regardless of whether data bytes changed:

828 // new tasks scheduled on this node, or lifecycle termination.

829 > func (n *Node) hasNewTransactionSideEffects() bool { tree.go

830 > return len(n.newTasks[n.value]) > 0 || n.terminated

831 > }

832

833 // serialize sets or updates serializedValue field of the node n with serialized value.

834 // It sets node's valueState to valueStateSynced and updates LastUpdateVersionedTransition.

835 > func (n *Node) serialize() error { tree.go

836 > switch n.serializedNode.GetMetadata().GetAttributes().(type) {

837 > case *persistencespb.ChasmNodeMetadata_ComponentAttributes: tree.go

838 > return n.serializeComponentNode()

839 case *persistencespb.ChasmNodeMetadata_DataAttributes:

840 return n.serializeDataNode()

841 case *persistencespb.ChasmNodeMetadata_CollectionAttributes:

842 return n.serializeCollectionNode()

843 > case *persistencespb.ChasmNodeMetadata_PointerAttributes: tree.go

844 > return n.serializePointerNode()

845 default:

846 return softassert.UnexpectedInternalErr(n.logger, "unknown node type", nil)

852 // LastUpdateVersionedTransition, the skip-if-clean revert logic in

853 // closeTransactionSerializeNodes must be updated accordingly.

854 > func (n *Node) serializeComponentNode() error { tree.go

855 > for field := range n.valueFields() {

856 > if field.err != nil {

857 return field.err

858 }

859

860 > if field.kind != fieldKindData { tree.go

861 > continue tree.go

862 }

863

864 > var blob *commonpb.DataBlob tree.go

865 > if !field.val.IsNil() {

866 > var err error tree.go

867 > if blob, err = encodeChasmBlob(field.val.Interface().(proto.Message)); err != nil {

868 return err

869 }

870 }

871

872 > n.serializedNode.Data = blob tree.go

873 >

874 > if n.serializedNode.GetMetadata().GetLastUpdateVersionedTransition() == nil {

875 > rc, ok := n.registry.componentFor(n.value) tree.go

876 > if !ok {

877 return softassert.UnexpectedInternalErr(

878 n.logger,

881 }

882 // TypeId mismatch on a brand new node indicates node reassignment.

883 > existingTypeID := n.serializedNode.GetMetadata().GetComponentAttributes().GetTypeId() tree.go

884 > if existingTypeID != 0 && existingTypeID != rc.componentID {

885 return softassert.UnexpectedInternalErr(

886 n.logger,

889 )

890 }

891 > n.serializedNode.GetMetadata().GetComponentAttributes().TypeId = rc.componentID tree.go

892 }

893

894 > n.updateLastUpdateVersionedTransition() tree.go

895 > n.setValueState(valueStateSynced)

896

897 // continue to iterate over fields to validate that there is only one proto field in the component.

898 }

899 > return nil tree.go

900 }

901

915 //

916 // nolint:revive,cognitive-complexity

917 > func (n *Node) syncSubComponents() error { tree.go

918 > if n.valueState < valueStateNeedSyncStructure {

919 for _, childNode := range n.children {

920 err := childNode.syncSubComponents()

926 }

927

928 > childrenToKeep := make(map[string]struct{}) tree.go

929 > for field := range n.valueFields() {

930 > if field.err != nil {

931 return field.err

932 }

933

934 > switch field.kind { tree.go

935 case fieldKindUnspecified:

936 softassert.Fail(n.logger,

937 "field.kind can be unspecified only if err is not nil, and there is a check for it above")

938 > case fieldKindData: tree.go

939 // Nothing to sync.

940 > case fieldKindSubField: tree.go

941 > keepChild, updatedFieldV, err := n.syncSubField(field.val, field.name)

942 > if err != nil {

943 return err

944 }

945 > if updatedFieldV.IsValid() { tree.go

946 > field.val.Set(updatedFieldV) tree.go

947 > }

948 > if keepChild { tree.go

949 > childrenToKeep[field.name] = struct{}{} tree.go

950 > }

951 > case fieldKindParentPtr: tree.go

952 > internalField := field.val.FieldByName(parentPtrInternalFieldName)

953 > internal, ok := internalField.Interface().(parentPtrInternal)

954 > if !ok {

955 return softassert.UnexpectedInternalErr(

956 n.logger,

958 fmt.Errorf("node %s, actual type: %T", n.nodeName, internalField.Interface()))

959 }

960 > if internal.currentNode == nil || internal.currentNode != n { tree.go

961 > internal.currentNode = n tree.go

962 > internalField.Set(reflect.ValueOf(internal))

963 > }

964 > case fieldKindSubMap: tree.go

965 > // Validate map type before doing anything with it.

966 > if !field.val.IsNil() && field.val.Kind() != reflect.Map {

967 return softassert.UnexpectedInternalErr(

968 n.logger,

971 }

972

973 > if field.val.IsNil() || len(field.val.MapKeys()) == 0 { tree.go

974 > // nil or empty map: skip without creating a collection node.

975 > // Any existing collection node will be removed by deleteChildren below.

976 > continue

977 }

978

1020 }

1021

1022 > err := n.deleteChildren(childrenToKeep) tree.go

1023 > n.setValueState(valueStateNeedSerialize)

1024 >

1025 > return err

1026 }

1027

1135 updatedFieldV reflect.Value,

1136 err error,

1137 > ) { tree.go

1138 > internalV := fieldV.FieldByName(internalFieldName)

1139 > //nolint:revive // Internal field is guaranteed to be of type fieldInternal.

1140 > internal := internalV.Interface().(fieldInternal)

1141 > if internal.isEmpty() {

1142 > // Internal is empty only when Field was explicitly set to NewEmptyField[T] which is a way to clear its value. tree.go

1143 > // In this case, return keepNode=false and this node (and all it children) will be added to DeletedNodes map.

1144 > return

1145 > }

1146

1147 > fieldValue := internal.value() tree.go

1148 > if internal.node == nil && fieldValue != nil {

1149 > fieldType := internal.fieldType() tree.go

1150 >

1151 > // Field is not empty but tree node is not set. It means this is a new field, and a node must be created.

1152 > childNode := newNode(n.nodeBase, n, fieldN)

1153 > childNode.initSerializedNode(fieldType)

1154 > childNode.setValueState(valueStateNeedSerialize)

1155 >

1156 > // set node value after validation

1157 > switch fieldType {

1158 > case fieldTypeComponent: tree.go

1159 > if err = assertStructPointer(reflect.TypeOf(fieldValue)); err != nil {

1160 return

1161 }

1162

1163 > childNode.setValueState(valueStateNeedSyncStructure) tree.go

1164 >

1165 > // Set detached flag from field option or component type registration.

1166 > componentAttr := childNode.serializedNode.GetMetadata().GetComponentAttributes()

1167 > componentAttr.Detached = internal.detached

1168 > if !componentAttr.Detached {

1169 > if rc, ok := n.registry.componentFor(fieldValue); ok {

1170 > componentAttr.Detached = rc.IsDetached()

1171 > }

1172 }

1173 case fieldTypeData:

1183 return

1184 }

1185 > case fieldTypeDeferredPointer: tree.go

1186 > n.needsPointerResolution = true

1187 default:

1188 err = softassert.UnexpectedInternalErr(

1193 return

1194 }

1195 > childNode.setValue(fieldValue) tree.go

1196 >

1197 > n.children[fieldN] = childNode

1198 > internal.node = childNode

1199 >

1200 > updatedFieldV = reflect.New(fieldV.Type()).Elem()

1201 > updatedFieldV.FieldByName(internalFieldName).Set(reflect.ValueOf(internal))

1202 }

1203

1204 > if internal.fieldType() == fieldTypeComponent && internal.value() != nil { tree.go

1205 > err = internal.node.syncSubComponents() tree.go

1206 > if err != nil {

1207 return

1208 }

1209 }

1210

1211 > return true, updatedFieldV, nil tree.go

1212 }

1213

1214 func (n *Node) deleteChildren(

1215 childrenToKeep map[string]struct{},

1216 > ) error { tree.go

1217 > for childName, childNode := range n.children {

1218 > if _, childToKeep := childrenToKeep[childName]; !childToKeep { tree.go

1219 if err := childNode.delete(false); err != nil {

1220 return err

1222 }

1223 }

1224 > return nil tree.go

1225 }

1226

1261

1262 // serializePointerNode doesn't serialize anything but named this way for consistency.

1263 > func (n *Node) serializePointerNode() error { tree.go

1264 > path, isPathValid := n.value.([]string)

1265 > if !isPathValid {

1266 return softassert.UnexpectedInternalErr(

1267 n.logger,

1270 }

1271

1272 > n.serializedNode.GetMetadata().GetPointerAttributes().NodePath = path tree.go

1273 > n.updateLastUpdateVersionedTransition()

1274 > n.setValueState(valueStateSynced)

1275 >

1276 > return nil

1277 }

1278

1279 > func (n *Node) updateLastUpdateVersionedTransition() { tree.go

1280 > if n.serializedNode.GetMetadata().GetLastUpdateVersionedTransition() == nil {

1281 > n.serializedNode.GetMetadata().LastUpdateVersionedTransition = &persistencespb.VersionedTransition{} tree.go

1282 > }

1283 > n.serializedNode.GetMetadata().GetLastUpdateVersionedTransition().TransitionCount = n.backend.NextTransitionCount() tree.go

1284 > n.serializedNode.GetMetadata().GetLastUpdateVersionedTransition().NamespaceFailoverVersion = n.backend.GetCurrentVersion()

1285 }

1286

1560 // reference a component that was never registered as a node are dropped and

1561 // logged at warn level to surface caller misuse.

1562 > func (n *Node) closeTransactionApplyPendingComponentMetadata() error { tree.go

1563 > if len(n.pendingRequestLinks) == 0 && len(n.pendingUserMetadata) == 0 {

1564 > return nil

1565 > }

1566 for _, node := range n.andAllChildren() {

1567 if !node.applyPendingComponentMetadata() {

1633 func (n *Node) componentNodePath(

1634 component Component,

1635 > ) ([]string, error) { tree.go

1636 > // It's unnecessary to deserialize entire tree as calling this method means

1637 > // caller already have the deserialized value.

1638 >

1639 > refNode, ok := n.valueToNode[component]

1640 > if !ok || !refNode.isComponent() {

1641 return nil, errComponentNotFound

1642 }

1643

1644 > return refNode.path(), nil tree.go

1645 }

1646

1663 func (n *Node) Now(

1664 _ Component,

1665 > ) time.Time { tree.go

1666 > // TODO: Now() could be different for components after we support Pause for CHASM components.

1667 > return n.timeSource.Now()

1668 > }

1669

1670 // AddTask implements the CHASM MutableContext interface

1692 // CloseTransaction is used by MutableState to close the transaction and

1693 // track changes made in the current transaction.

1694 > func (n *Node) CloseTransaction() (NodesMutation, error) { tree.go

1695 > defer n.cleanupTransaction()

1696 >

1697 > if err := n.executeImmediatePureTasks(); err != nil {

1698 return NodesMutation{}, err

1699 }

1700

1701 > if err := n.syncSubComponents(); err != nil { tree.go

1702 return NodesMutation{}, err

1703 }

1704

1705 > if n.needsPointerResolution { tree.go

1706 > if err := n.resolveDeferredPointers(); err != nil { tree.go

1707 return NodesMutation{}, err

1708 }

1709 }

1710

1711 > nextVersionedTransition := &persistencespb.VersionedTransition{ tree.go

1712 > NamespaceFailoverVersion: n.backend.GetCurrentVersion(),

1713 > TransitionCount: n.backend.NextTransitionCount(),

1714 > }

1715 >

1716 > immutableContext := NewContext(context.TODO(), n)

1717 > rootLifecycleChanged, err := n.closeTransactionHandleRootLifecycleChange(immutableContext)

1718 > if err != nil {

1719 return NodesMutation{}, err

1720 }

1721

1722 > if n.subtreeIsDirty { tree.go

1723 > if err := n.closeTransactionForceUpdateVisibility(immutableContext, rootLifecycleChanged); err != nil { tree.go

1724 return NodesMutation{}, err

1725 }

1726 }

1727

1728 > if err := n.closeTransactionSerializeNodes(); err != nil { tree.go

1729 return NodesMutation{}, err

1730 }

1731

1732 > if err := n.closeTransactionUpdateComponentTasks(nextVersionedTransition); err != nil { tree.go

1733 return NodesMutation{}, err

1734 }

1735

1736 > if err := n.closeTransactionApplyPendingComponentMetadata(); err != nil { tree.go

1737 return NodesMutation{}, err

1738 }

1739

1740 // Both user & system data mutation need to be returned and persisted.

1741 > maps.Copy(n.mutation.UpdatedNodes, n.systemMutation.UpdatedNodes) tree.go

1742 > maps.Copy(n.mutation.DeletedNodes, n.systemMutation.DeletedNodes)

1743 >

1744 > return n.mutation, nil

1745 }

1746

1747 > func (n *Node) executeImmediatePureTasks() error { tree.go

1748 >

1749 > // We must sync structure before running any tasks here because,

1750 > // those tasks might be for a newly created component which doesn't even have a node yet.

1751 > // And we want to make sure we only run tasks for components that are still part of the tree.

1752 > syncStructure := true

1753 > var err error

1754 >

1755 > for len(n.immediatePureTasks) != 0 {

1756 // Create a map in case more immediate pure tasks get

1757 // added while existing ones are executed.

1786 }

1787

1788 > return nil tree.go

1789 }

1790

1791 func (n *Node) closeTransactionHandleRootLifecycleChange(

1792 immutableContext Context,

1793 > ) (bool, error) { tree.go

1794 > if n.backend.IsWorkflow() {

1795 // Workflow manages its lifecycle directly in mutable state.

1796 return false, nil

1797 }

1798

1799 > if n.valueState != valueStateNeedSerialize { tree.go

1800 return false, nil

1801 }

1802

1803 > if n.backend.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED { tree.go

1804 // Already in completed state, no need to update lifecycle state.

1805 return false, nil

1806 }

1807

1808 > if n.terminated { tree.go

1809 return n.backend.UpdateWorkflowStateStatus(

1810 enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,

1813 }

1814

1815 > rootComponent, err := n.Component(immutableContext, ComponentRef{}) tree.go

1816 > if err != nil {

1817 return false, err

1818 }

1819 > lifecycleState := rootComponent.LifecycleState(immutableContext) tree.go

1820 >

1821 > var newState enumsspb.WorkflowExecutionState

1822 > var newStatus enumspb.WorkflowExecutionStatus

1823 > switch lifecycleState {

1824 > case LifecycleStateRunning, LifecycleStatePaused:

1825 > // Paused is an OPEN state; the execution remains RUNNING from the persistence perspective.

1826 > newState = enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING

1827 > newStatus = enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING

1828 case LifecycleStateCompleted:

1829 newState = enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED

1839 }

1840

1841 > return n.backend.UpdateWorkflowStateStatus(newState, newStatus) tree.go

1842 }

1843

1845 immutableContext Context,

1846 rootLifecycleChanged bool,

1847 > ) error { tree.go

1848 > if n.deleteAfterClose {

1849 return nil

1850 }

1851

1852 > if !rootLifecycleChanged && tree.go

1853 > n.backend.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {

1854 return nil

1855 }

1856

1857 > needUpdate := rootLifecycleChanged tree.go

1858 >

1859 > rootComponent, err := n.Component(immutableContext, ComponentRef{})

1860 > if err != nil {

1861 return err

1862 }

1863

1864 > saProvider, ok := rootComponent.(VisibilitySearchAttributesProvider) tree.go

1865 > if ok {

1866 > saSlice := saProvider.SearchAttributes(immutableContext) tree.go

1867 > newSA := searchAttributeKeyValuesToMap(saSlice)

1868 > if !maps.EqualFunc(n.currentSA, newSA, isVisibilityValueEqual) {

1869 > needUpdate = true tree.go

1870 > }

1871 > n.currentSA = newSA tree.go

1872 }

1873

1874 > memoProvider, ok := rootComponent.(VisibilityMemoProvider) tree.go

1875 > if ok {

1876 > newMemo := memoProvider.Memo(immutableContext) tree.go

1877 > if !proto.Equal(n.currentMemo, newMemo) {

1878 > needUpdate = true tree.go

1879 > }

1880 > n.currentMemo = proto.Clone(newMemo) tree.go

1881 }

1882

1883 > if !needUpdate { tree.go

1884 > return nil tree.go

1885 > }

1886

1887 > var visibilityNode *Node tree.go

1888 > for _, child := range n.children {

1889 > if !child.isComponent() { tree.go

1890 continue

1891 }

1892

1893 > if child.valueState == valueStateNeedSerialize { tree.go

1894 > if rc, ok := n.registry.componentFor(child.value); ok && rc.fqType() == visibilityComponentType { tree.go

1895 visibilityNode = child

1896 break

1902 }

1903

1904 > if visibilityNode == nil { tree.go

1905 > return nil tree.go

1906 > }

1907

1908 visComponent, err := visibilityNode.Component(immutableContext, ComponentRef{})

1933 }

1934

1935 > func (n *Node) closeTransactionSerializeNodes() error { tree.go

1936 > for nodePath, node := range n.andAllChildren() {

1937 > if node.valueState > valueStateNeedSerialize {

1938 return serviceerror.NewInternalf("invalid valueState for serializing: %v", node.valueState)

1939 }

1940

1941 > if node.valueState < valueStateNeedSerialize { tree.go

1942 > continue tree.go

1943 }

1944

1945 > encodedPath, err := node.getEncodedPath() tree.go

1946 > if err != nil {

1947 return err

1948 }

1952 // prevData captures the pre-serialize blob pointer; serialize() allocates a new

1953 // blob, leaving prevData pointing at the original for comparison.

1954 > prevVersionedTransition := common.CloneProto( tree.go

1955 > node.serializedNode.GetMetadata().GetLastUpdateVersionedTransition(),

1956 > )

1957 > skipIfClean := (node.isComponent() || node.isData() || node.isMap()) &&

1958 > prevVersionedTransition != nil &&

1959 > !node.hasNewTransactionSideEffects()

1960 > var prevData *commonpb.DataBlob

1961 > if skipIfClean {

1962 > prevData = node.serializedNode.Data tree.go

1963 > }

1964

1965 > if err := node.serialize(); err != nil { tree.go

1966 return err

1967 }

1968

1969 // Data bytes unchanged: revert the versioned transition bump and skip persistence.

1970 > if skipIfClean && bytes.Equal(prevData.GetData(), node.serializedNode.Data.GetData()) { tree.go

1971 > node.serializedNode.GetMetadata().LastUpdateVersionedTransition = prevVersionedTransition tree.go

1972 > continue

1973 }

1974

1975 > if componentAttr := node.serializedNode.GetMetadata().GetComponentAttributes(); componentAttr != nil && tree.go

1976 > componentAttr.TypeId == visibilityComponentTypeID &&

1977 > len(nodePath) != 1 {

1978 return softassert.UnexpectedInternalErr(

1979 n.logger,

1982 }

1983

1984 > n.mutation.UpdatedNodes[encodedPath] = node.serializedNode tree.go

1985 > // DeletedNodes map is populated when syncing tree structure. However, since we may sync tree structure

1986 > // multiple times in one transaction, if node at the same path was previously deleted, have structure synced,

1987 > // then get re-created, the same encoded path will exists in both UpdatedNodes and DeletedNodes maps.

1988 > //

1989 > // serializeNode only happens once at the end of a transaction, and here we know the node at this encoded path exists,

1990 > // remove it from the DeletedNodes map.

1991 > delete(n.mutation.DeletedNodes, encodedPath)

1992 }

1993

1994 > return nil tree.go

1995 }

1996

1997 func (n *Node) closeTransactionUpdateComponentTasks(

1998 nextVersionedTransition *persistencespb.VersionedTransition,

1999 > ) error { tree.go

2000 > taskOffset := int64(1)

2001 > taskValidationContext := NewContext(newContextWithOperationIntent(context.Background(), OperationIntentProgress), n)

2002 >

2003 > archetypeID := n.ArchetypeID()

2004 >

2005 > var firstPureTask *persistencespb.ChasmComponentAttributes_Task

2006 > var firstPureTaskNode *Node

2007 >

2008 > for nodePath, node := range n.andAllChildren() {

2009 > // no-op if node is not a component

2010 > componentAttr := node.serializedNode.Metadata.GetComponentAttributes()

2011 > if componentAttr == nil {

2012 > continue tree.go

2013 }

2014

2021 // markSubtreeDirty propagates to both ancestors and descendants at mutation time,

2022 // so we skip validation only for nodes with no dirty node anywhere in their lineage.

2023 > if node.subtreeIsDirty { tree.go

2024 > // Ensure this node's component value is hydrated before cleaning up tasks. tree.go

2025 > if err := node.prepareComponentValue(taskValidationContext); err != nil {

2026 return err

2027 }

2028

2029 > cleanedUp, err := node.closeTransactionCleanupInvalidTasks(taskValidationContext) tree.go

2030 > if err != nil {

2031 return err

2032 }

2033

2034 > if cleanedUp { tree.go

2035 // add the current node to UpdatedNodes map if it's not already there

2036 encodedPath, err := node.getEncodedPath()

2054 // This method is called after the closeTransactionSerializeNodes which sets valueState

2055 // to valueStateSynced.

2056 > if transitionhistory.Compare( tree.go

2057 > node.serializedNode.GetMetadata().LastUpdateVersionedTransition,

2058 > nextVersionedTransition,

2059 > ) == 0 && node.valueState != valueStateNeedDeserialize {

2060 > if err := node.closeTransactionHandleNewTasks( tree.go

2061 > nextVersionedTransition,

2062 > taskValidationContext,

2063 > &taskOffset,

2064 > ); err != nil {

2065 return err

2066 }

2067 }

2068

2069 > sideEffectTasks := componentAttr.GetSideEffectTasks() tree.go

2070 > for idx := len(sideEffectTasks) - 1; idx >= 0; idx-- {

2071 sideEffectTask := sideEffectTasks[idx]

2072 if sideEffectTask.PhysicalTaskStatus == physicalTaskStatusCreated {

2083 // Find the first pure task in the entire tree,

2084 // regardless if the pure task is newly added or existing.

2085 > pureTasks := componentAttr.GetPureTasks() tree.go

2086 > if len(pureTasks) == 0 {

2087 > continue tree.go

2088 }

2089

2101 // task, and that component get deleted by the second task.

2102

2103 > return n.closeTransactionGeneratePhysicalPureTask( tree.go

2104 > firstPureTask,

2105 > firstPureTaskNode,

2106 > archetypeID,

2107 > )

2108 }

2109

2164 func (n *Node) closeTransactionCleanupInvalidTasks(

2165 validateContext Context,

2166 > ) (bool, error) { tree.go

2167 > // Validate existing tasks and remove invalid ones.

2168 > var validationErr error

2169 > cleanedUp := false

2170 > deleteFunc := func(existingTask *persistencespb.ChasmComponentAttributes_Task) bool {

2171 existingTaskInstance, err := n.deserializeComponentTask(existingTask)

2172 if err != nil {

2196 }

2197

2198 > componentAttr := n.serializedNode.Metadata.GetComponentAttributes() tree.go

2199 > componentAttr.SideEffectTasks = slices.DeleteFunc(componentAttr.SideEffectTasks, deleteFunc)

2200 > if validationErr != nil {

2201 return false, validationErr

2202 }

2203 > componentAttr.PureTasks = slices.DeleteFunc(componentAttr.PureTasks, deleteFunc) tree.go

2204 > if validationErr != nil {

2205 return false, validationErr

2206 }

2207 > return cleanedUp, nil tree.go

2208 }

2209

2242 validateContext Context,

2243 taskOffset *int64,

2244 > ) error { tree.go

2245 > newTasks, ok := n.newTasks[n.value]

2246 > if !ok {

2247 > return nil tree.go

2248 > }

2249

2250 componentAttr := n.serializedNode.Metadata.GetComponentAttributes()

2346 firstTaskNode *Node,

2347 archetypeID ArchetypeID,

2348 > ) error { tree.go

2349 > if firstPureTask == nil {

2350 > n.backend.DeleteCHASMPureTasks(tasks.MaximumKey.FireTime) tree.go

2351 > return nil

2352 > }

2353

2354 firstPureTaskScheduledTime := firstPureTask.ScheduledTime.AsTime()

2381 // Returns error if any deferred pointer cannot be resolved, as deferred pointers

2382 // cannot be persisted after transaction close.

2383 > func (n *Node) resolveDeferredPointers() error { tree.go

2384 > for _, node := range n.andAllChildren() {

2385 > if node.value == nil || !node.isComponent() {

2386 > continue tree.go

2387 }

2388

2389 > for field := range node.valueFields() { tree.go

2390 > if field.err != nil {

2391 return field.err

2392 }

2393

2394 > if field.kind != fieldKindSubField { tree.go

2395 > continue

2396 }

2397

2398 > internalV := field.val.FieldByName(internalFieldName) tree.go

2399 > internal, _ := internalV.Interface().(fieldInternal) //nolint:revive

2400 >

2401 > if internal.fieldType() == fieldTypeDeferredPointer && internal.value() != nil {

2402 > // Must resolve the deferred pointer or fail the transaction.

2403 > var resolvedPath []string

2404 > var err error

2405 >

2406 > switch value := internal.value().(type) {

2407 > case Component:

2408 > resolvedPath, err = n.componentNodePath(value)

2409 > if err == nil {

2410 > targetNode := n.valueToNode[value] tree.go

2411 > if !targetNode.isAncestorOf(node) {

2412 err = fmt.Errorf(

2413 "pointer target is not an ancestor of component at path %v",

2424 fmt.Errorf("%T", value))

2425 }

2426 > if err != nil { tree.go

2427 return softassert.UnexpectedInternalErr(

2428 n.logger,

2433 // Update the field to be a regular pointer, reusing the existing serializedNode,

2434 // and update the serializedNode's value.

2435 > newInternal := newFieldInternalWithValue(fieldTypePointer, resolvedPath) tree.go

2436 > newInternal.node = internal.node

2437 > newInternal.node.setValue(resolvedPath)

2438 > internalV.Set(reflect.ValueOf(newInternal))

2439 }

2440 }

2441 }

2442 > return nil tree.go

2443 }

2444

2445 // andAllChildren returns a sequence of all nodes in the tree starting from n, including n itself.

2446 // The sequence is depth-first, pre-order traversal.

2447 > func (n *Node) andAllChildren() iter.Seq2[[]string, *Node] { tree.go

2448 > return func(yield func([]string, *Node) bool) {

2449 > var walk func([]string, *Node) bool

2450 > walk = func(path []string, node *Node) bool {

2451 > if node == nil {

2452 return true

2453 }

2454 > if !yield(path, node) { tree.go

2455 return false

2456 }

2457 > for _, child := range node.children { tree.go

2458 > childPath := make([]string, len(path)+1) tree.go

2459 > copy(childPath, path)

2460 > childPath[len(path)] = child.nodeName

2461 > if !walk(childPath, child) {

2462 return false

2463 }

2464 }

2465 > return true tree.go

2466 }

2467 > walk(nil, n) tree.go

2468 }

2469 }

2470

2471 > func (n *Node) cleanupTransaction() { tree.go

2472 > n.mutation = NodesMutation{

2473 > UpdatedNodes: make(map[string]*persistencespb.ChasmNode),

2474 > DeletedNodes: make(map[string]struct{}),

2475 > }

2476 >

2477 > // System mutation are most likely to be empty, so we reuse existing ones if possible.

2478 > if len(n.systemMutation.UpdatedNodes) != 0 {

2479 n.systemMutation.UpdatedNodes = make(map[string]*persistencespb.ChasmNode)

2480 }

2481 > if len(n.systemMutation.DeletedNodes) != 0 { tree.go

2482 n.systemMutation.DeletedNodes = make(map[string]struct{})

2483 }

2484

2485 > n.newTasks = make(map[any][]taskWithAttributes) tree.go

2486 > if len(n.immediatePureTasks) != 0 {

2487 // n.immediatePureTasks should already be empty after executeImmediatePureTasks()

2488 // unless there's an error.

2490 }

2491

2492 > if len(n.pendingRequestLinks) != 0 { tree.go

2493 n.pendingRequestLinks = make(map[any]map[string][]*commonpb.Link)

2494 }

2495 > if len(n.pendingUserMetadata) != 0 { tree.go

2496 n.pendingUserMetadata = make(map[any]*sdkpb.UserMetadata)

2497 }

2498

2499 > n.needsPointerResolution = false tree.go

2500 >

2501 > // Reset per-node subtreeIsDirty on all nodes in the tree.

2502 > for _, node := range n.andAllChildren() {

2503 > node.subtreeIsDirty = false

2504 > }

2505 }

2506

2931 }

2932

2933 > func (n *Node) getEncodedPath() (string, error) { tree.go

2934 > if n.encodedPath != nil {

2935 > return *n.encodedPath, nil tree.go

2936 > }

2937 > encodePath, err := n.pathEncoder.Encode(n, n.path()) tree.go

2938 > if err == nil {

2939 > n.encodedPath = &encodePath

2940 > }

2941 > return encodePath, err

2942 }

2943

2944 > func (n *Node) path() []string { tree.go

2945 > if n.parent == nil {

2946 > return []string{}

2947 > }

2948

2949 > return append(n.parent.path(), n.nodeName) tree.go

2950 }

2951

2952 func (n *Node) findNode(

2953 path []string,

2954 > ) (*Node, bool) { tree.go

2955 > if len(path) == 0 {

2956 > return n, true

2957 > }

2958

2959 childName := path[0]

2967 // isAncestorOf returns true if n is a proper ancestor of descendant.

2968 // It walks from descendant up through parent links to check if n is encountered.

2969 > func (n *Node) isAncestorOf(descendant *Node) bool { tree.go

2970 > current := descendant.parent

2971 > for current != nil {

2972 > if current == n { tree.go

2973 > return true

2974 > }

2975 > current = current.parent tree.go

2976 }

2977 return false

3115

3116 // ArchetypeID returns the framework's internal ID for the root component's fully qualified name.

3117 > func (n *Node) ArchetypeID() ArchetypeID { tree.go

3118 > // Root must be a component.

3119 > return n.root().serializedNode.Metadata.GetComponentAttributes().GetTypeId()

3120 > }

3121

3122 // Archetype returns the root component's fully qualified name.

3136 }

3137

3138 > func (n *Node) root() *Node { tree.go

3139 > if n.parent == nil {

3140 > return n

3141 > }

3142 > return n.parent.root() tree.go

3143 }

3144

3261 parent *Node,

3262 nodeName string,

3263 > ) *Node { tree.go

3264 > return &Node{

3265 > nodeBase: base,

3266 > parent: parent,

3267 > children: make(map[string]*Node),

3268 > nodeName: nodeName,

3269 > }

3270 > }

3271

3272 func compareSideEffectTasks(a, b *persistencespb.ChasmComponentAttributes_Task) int {

3868 // encodeChasmBlob encodes CHASM data and task payloads through the env-aware

3869 // serializer while preserving deterministic proto3 bytes for byte comparisons.

3870 > func encodeChasmBlob(m proto.Message) (*commonpb.DataBlob, error) { tree.go

3871 > return serialization.Encode(m, serialization.WithDeterministicProto3)

3872 > }