go.temporal.io/server/chasm/tree.go
3872 LOC · 1873 covered · 1999 uncovered · 624 ranges · 1038 concepts · 283 introducers · 384 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package chasm
2
3
import (
4
"bytes"
5
"cmp"
6
"context"
7
"errors"
8
"fmt"
9
"iter"
10
"reflect"
11
"slices"
12
"strconv"
13
"time"
14
15
commonpb "go.temporal.io/api/common/v1"
16
enumspb "go.temporal.io/api/enums/v1"
17
historypb "go.temporal.io/api/history/v1"
18
sdkpb "go.temporal.io/api/sdk/v1"
19
"go.temporal.io/api/serviceerror"
20
enumsspb "go.temporal.io/server/api/enums/v1"
21
persistencespb "go.temporal.io/server/api/persistence/v1"
22
"go.temporal.io/server/common"
23
"go.temporal.io/server/common/clock"
24
"go.temporal.io/server/common/definition"
25
"go.temporal.io/server/common/log"
26
"go.temporal.io/server/common/log/tag"
27
"go.temporal.io/server/common/metrics"
28
"go.temporal.io/server/common/namespace"
29
"go.temporal.io/server/common/nexus/nexusrpc"
30
"go.temporal.io/server/common/persistence/serialization"
31
"go.temporal.io/server/common/persistence/transitionhistory"
32
"go.temporal.io/server/common/softassert"
33
"go.temporal.io/server/service/history/tasks"
34
"golang.org/x/exp/maps"
35
"google.golang.org/protobuf/proto"
36
"google.golang.org/protobuf/types/known/timestamppb"
37
)
38
39
var (
40
protoMessageT = reflect.TypeFor[proto.Message]()
41
)
42
43
var (
44
errAccessCheckFailed = serviceerror.NewNotFound("access check failed, CHASM tree is closed for writes")
45
errComponentNotFound = serviceerror.NewNotFound("component not found")
46
errDataNotFound = serviceerror.NewNotFound("data not found")
47
errTaskNotValid = serviceerror.NewNotFound("task is no longer valid")
48
)
49
50
// valueState is an in-memory indicator of the dirtiness of a deserialized node value.
51
// The dirtiness has two parts:
52
// 1. If the data part of the value is in sync with the serializedNode field.
53
// 2. For component node, if the structure of the component is in sync with the children field.
54
//
55
// The enum value below is defined in increasing order of "dirtiness".
56
// - NeedDeserialize: Value is not even deserialized yet.
57
// - Synced: Value is deserialized and in sync with both serializedNode and children.
58
// - NeedSerialize: Value is deserialized, the child tree structure is synced, but the value is not in sync with serializedNode.
59
// - NeedSyncStructure: Value is deserialized, neither data nor tree structure is synced.
60
//
61
// For simplicity, for a dirty component node, the logic always sync structure (potentially multiple times within a transaction) first,
62
// and the serialize the data at the very end of a transaction. So there will never base a case where value is synced with seralizedNode,
63
// but not with children.
64
//
65
// To update this field, ALWAYS use setValueState() method.
66
//
67
// NOTE: This is a different concept from the IsDirty() method which is needed by MutableState implementation to determine
68
// if the state in memory matches the state in DB.
69
type valueState uint8
70
71
const (
72
valueStateUndefined valueState = iota
73
valueStateNeedDeserialize
74
valueStateSynced
75
valueStateNeedSerialize
76
valueStateNeedSyncStructure
77
)
78
79
const (
80
physicalTaskStatusNone int32 = iota
81
physicalTaskStatusCreated
82
)
83
84
type (
85
// Node is the in-memory representation of a persisted CHASM node.
86
//
87
// Node and all its methods are NOT meant to be used by CHASM component authors.
88
// They are exported for use by the CHASM engine and underlying MutableState implementation only.
89
Node struct {
90
*nodeBase
91
92
parent *Node
93
children map[string]*Node // child name (path segment) -> child node
94
nodeName string // key of this node in parent's children map, empty string for root node.
95
96
// Type of attributes controls the type of the node.
97
serializedNode *persistencespb.ChasmNode // serialized component | data | collection with metadata
98
// Deserialized component | data | map
99
// Do NOT set this field directly, use setValue() method instead.
100
value any
101
// Do NOT set this field directly, use setValueState() method instead.
102
valueState valueState
103
104
// Cached encoded path for this node.
105
// DO NOT read this field directly. Always use getEncodedPath() method to retrieve the encoded path.
106
//
107
// Empty string is a valid encoded path (for root node), so using *string here to differentiate.
108
//
109
// TODO: Consider using unique package here.
110
// Encoded path for different runs of the same Component type are the same.
111
encodedPath *string
112
113
// When terminated is true, regardless of the Lifecycle state of the component,
114
// the component will be considered as closed.
115
//
116
// NOTE: this is an in-memory only field and will be lost upon mutable state reload or replication.
117
// The purpose of this field is only for the transaction that force terminates the execution to
118
// update executionState & State in mutable state and generate retention timers, so it only needs to be
119
// in-memory and on the active side.
120
// If your logic needs to check if an execution is ever force terminated, check both this field (for the current
121
// transaction) and also the executionState from backend (for previous transactions).
122
//
123
// We can consider extending the force terminate concept to sub-components as well, and make the field durable.
124
terminated bool
125
126
// deleteAfterClose suppresses the close visibility task when an execution is being
127
// terminated as part of a delete operation. Like terminated, this is in-memory only
128
// and only needed for the current transaction. Set via SetDeleteAfterClose.
129
deleteAfterClose bool
130
131
// subtreeIsDirty is true if this node, any ancestor, or any descendant was mutated
132
// in the current transaction (valueState >= valueStateNeedSerialize), or if
133
// ExecutePureTask ran on this node or any such relative.
134
//
135
// markSubtreeDirty propagates the flag both upward (to ancestors) and downward (to
136
// all descendants) at mutation time, so CloseTransaction can skip task validation
137
// for nodes whose entire lineage is clean with a single O(1) flag check.
138
//
139
// This is a per-node field (not in nodeBase) and is reset after each transaction.
140
subtreeIsDirty bool
141
}
142
143
// nodeBase is a set of dependencies and states shared by all nodes in a CHASM tree.
144
nodeBase struct {
145
registry *Registry
146
timeSource clock.TimeSource
147
backend NodeBackend
148
pathEncoder NodePathEncoder
149
logger log.Logger
150
metricsHandler metrics.Handler
151
152
// Following fields are changes accumulated in this transaction,
153
// and will get cleaned up after CloseTransaction().
154
155
// mutation field captures all user state changes (those will be replicated)
156
mutation NodesMutation
157
// systemMutation field captures all cell specific system changes (those will NOT be replicated)
158
systemMutation NodesMutation
159
160
newTasks map[any][]taskWithAttributes // component value -> task & attributes
161
immediatePureTasks map[any][]taskWithAttributes // similar to newTasks, but will be executed at the end of the transaction
162
163
// Pending framework metadata writes keyed by component value. Applied to
164
// each component's ChasmComponentAttributes during CloseTransaction so
165
// callers can stage writes before the component is registered as a node.
166
pendingRequestLinks map[any]map[string][]*commonpb.Link
167
pendingUserMetadata map[any]*sdkpb.UserMetadata
168
169
// Node value -> node
170
// Only component and data node values are tracked right now
171
valueToNode map[any]*Node
172
173
taskValueCache map[*commonpb.DataBlob]reflect.Value
174
175
// Root component's search attributes and memo at the start of a transaction.
176
// They will be updated upon CloseTransaction() if they are changed.
177
currentSA map[string]VisibilityValue
178
currentMemo proto.Message
179
180
needsPointerResolution bool
181
}
182
183
taskWithAttributes struct {
184
task any
185
attributes TaskAttributes
186
}
187
188
// NodesMutation is a set of mutations for all nodes rooted at a given node n,
189
// including the node n itself.
190
NodesMutation struct {
191
UpdatedNodes map[string]*persistencespb.ChasmNode // encoded node path -> chasm node
192
DeletedNodes map[string]struct{}
193
}
194
195
// NodesSnapshot is a snapshot for all nodes rooted at a given node n,
196
// including the node n itself.
197
NodesSnapshot struct {
198
Nodes map[string]*persistencespb.ChasmNode // encoded node path -> chasm node
199
}
200
201
// NodeBackend is a set of methods needed from MutableState.
202
//
203
// This is for breaking cycle dependency between
204
// this package and service/history/workflow package
205
// where MutableState is defined.
206
NodeBackend interface {
207
// TODO: Add methods needed from MutateState here.
208
GetExecutionState() *persistencespb.WorkflowExecutionState
209
GetExecutionInfo() *persistencespb.WorkflowExecutionInfo
210
GetApproximatePersistedSize() int
211
GetNamespaceEntry() *namespace.Namespace
212
GetCurrentVersion() int64
213
NextTransitionCount() int64
214
CurrentVersionedTransition() *persistencespb.VersionedTransition
215
GetWorkflowKey() definition.WorkflowKey
216
AddTasks(...tasks.Task)
217
AddHistoryEvent(t enumspb.EventType, setAttributes func(*historypb.HistoryEvent)) *historypb.HistoryEvent
218
GenerateEventLoadToken(event *historypb.HistoryEvent) ([]byte, error)
219
LoadHistoryEvent(ctx context.Context, token []byte) (*historypb.HistoryEvent, error)
220
HasAnyBufferedEvent(filter func(*historypb.HistoryEvent) bool) bool
221
DeleteCHASMPureTasks(maxScheduledTime time.Time)
222
UpdateWorkflowStateStatus(
223
state enumsspb.WorkflowExecutionState,
224
status enumspb.WorkflowExecutionStatus,
225
) (bool, error)
226
IsWorkflow() bool
227
GetNexusCompletion(
228
ctx context.Context,
229
requestID string,
230
) (nexusrpc.CompleteOperationOptions, error)
231
GetNexusUpdateCompletion(
232
ctx context.Context,
233
updateID string,
234
requestID string,
235
) (nexusrpc.CompleteOperationOptions, error)
236
EndpointRegistry() EndpointRegistry
237
}
238
239
// NodePathEncoder is an interface for encoding and decoding node paths.
240
// Logic outside the chasm package should only work with encoded paths.
241
NodePathEncoder interface {
242
Encode(node *Node, path []string) (string, error)
243
// TODO: Return a iterator on node name instead of []string,
244
// so that we can get a node by encoded path without additional
245
// allocation for the decoded path.
246
Decode(encodedPath string) ([]string, error)
247
}
248
249
// NodePureTask is intended to be implemented and used within the CHASM
250
// framework only.
251
NodePureTask interface {
252
ExecutePureTask(baseCtx context.Context, taskAttributes TaskAttributes, taskInstance any) (bool, error)
253
}
254
)
255
256
// IsEmpty reports whether the mutation contains no node updates or deletions.
258
>
return len(m.UpdatedNodes) == 0 && len(m.DeletedNodes) == 0
259
>
}
260
261
// NewTreeFromDB creates a new in-memory CHASM tree from a collection of flattened persistence CHASM nodes.
262
// This method should only be used when loading an existing CHASM tree from database.
263
// If serializedNodes is empty, the tree will be considered as a legacy Workflow execution without any CHASM nodes.
264
func NewTreeFromDB(
265
serializedNodes map[string]*persistencespb.ChasmNode, // This is coming from MS map[nodePath]ChasmNode.
266
registry *Registry,
267
timeSource clock.TimeSource,
268
backend NodeBackend,
269
pathEncoder NodePathEncoder,
270
logger log.Logger,
271
metricsHandler metrics.Handler,
273
>
if len(serializedNodes) == 0 {
274
>
root := NewEmptyTree(registry, timeSource, backend, pathEncoder, logger, metricsHandler)
tree.go ×1
275
>
// NewEmptyTree initializes the serializedNode to an empty component node,
276
>
root.serializedNode.Metadata.GetComponentAttributes().TypeId = WorkflowArchetypeID
277
>
return root, nil
278
>
}
279
280
>
root := newTreeHelper(registry, timeSource, backend, pathEncoder, logger, metricsHandler)
tree.go ×9
281
>
for encodedPath, serializedNode := range serializedNodes {
282
>
nodePath, err := pathEncoder.Decode(encodedPath)
283
>
if err != nil {
284
return nil, err
285
}
287
}
288
290
return nil, err
291
}
293
}
294
295
// NewEmptyTree creates a new empty in-memory CHASM tree.
296
func NewEmptyTree(
297
registry *Registry,
298
timeSource clock.TimeSource,
299
backend NodeBackend,
300
pathEncoder NodePathEncoder,
301
logger log.Logger,
302
metricsHandler metrics.Handler,
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(
319
registry *Registry,
320
timeSource clock.TimeSource,
321
backend NodeBackend,
322
pathEncoder NodePathEncoder,
323
logger log.Logger,
324
metricsHandler metrics.Handler,
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(
355
root *Node,
356
registry *Registry,
358
>
immutableContext := NewContext(context.Background(), root)
359
>
rootComponent, err := root.Component(immutableContext, ComponentRef{})
360
>
if err != nil {
361
return err
362
}
363
364
// Theoritically we should check if the root node has a Visibility component or not.
365
// But that doesn't really matter. Even if it doesn't have one, currentSearchAttributes
366
// and currentMemo will just never be used.
367
370
>
root.currentSA = searchAttributeKeyValuesToMap(saSlice)
371
>
}
374
>
}
375
377
}
378
379
>
func searchAttributeKeyValuesToMap(saSlice []SearchAttributeKeyValue) map[string]VisibilityValue {
tree.go ×1
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,
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
399
// setValue sets the value field of the node.
400
// If the node is a component or data node, the index from node value to node (valueToNode)
401
// is also updated.
403
>
if !n.isComponent() && !n.isData() {
405
>
return
406
>
}
407
410
>
}
411
413
>
414
>
if value != nil {
415
>
n.valueToNode[value] = n
416
>
}
417
}
418
420
>
n.valueState = state
421
>
if state >= valueStateNeedSerialize {
423
>
}
424
}
425
426
// markSubtreeDirty sets subtreeIsDirty on this node and propagates upward to all ancestors.
427
// This ensures that ancestor nodes know their subtree contains a dirty node, which is used
428
// during CloseTransaction to skip task validation for completely unrelated subtrees.
429
// markSubtreeDirty marks this node and its entire lineage (ancestors and descendants)
430
// as dirty so that CloseTransaction knows to validate their tasks.
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.
438
>
desc.subtreeIsDirty = true
439
>
}
440
}
441
443
>
for node := parent; node != nil; node = node.parent {
446
}
447
449
>
node.setValueState(valueStateNeedDeserialize)
450
}
451
}
452
453
// Component retrieves a component from the tree rooted at node n
454
// using the provided component reference
455
// It also performs access rule, and task validation checks
456
// (for task processing requests) before returning the component.
457
func (n *Node) Component(
458
chasmContext Context,
459
ref ComponentRef,
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 {
467
>
}
468
470
>
ref.componentInitialVT,
471
>
node.serializedNode.Metadata.InitialVersionedTransition,
472
>
) != 0 {
474
>
}
475
477
>
if err := node.prepareComponentValue(validationContext); err != nil {
478
return nil, err
479
}
480
482
>
if !ok {
483
return nil, softassert.UnexpectedInternalErr(
484
n.logger,
485
"component value is not of type Component",
486
fmt.Errorf("%s", reflect.TypeOf(node.value).String()))
487
}
488
490
return nil, err
491
}
492
494
>
if err := ref.validationFn(node.root().backend, validationContext, componentValue, node.registry); err != nil {
tree.go ×1
496
>
}
497
}
498
499
// prepare component value again using incoming context to mark node as dirty if needed.
501
return nil, err
502
}
504
}
505
506
// validateAccess performs the access rule check on a node.
507
//
508
// When the context's intent is OperationIntentProgress, This check validates that
509
// all of a node's ancestors are still in a running state, and can accept writes. In
510
// the case of a newly created node, a detached node, or an OperationIntentObserve
511
// intent, the check is skipped.
512
//
513
// When checkPaused is true (used during task validation), the check is extended to
514
// also treat a paused lifecycle state as a blocking condition - for both ancestors
515
// and the node itself. This collapses the paused-subtree traversal into the same
516
// single pass, avoiding a second tree walk.
517
// Note: engine mutations on paused components are still accepted (checkPaused=false),
518
// per the current requirement.
520
>
intent := operationIntentFromContext(ctx.goContext())
521
>
if intent != OperationIntentProgress {
523
>
return nil
524
>
}
525
526
// Detached nodes skip ancestor validation entirely.
529
>
}
530
534
>
}
535
}
536
537
// validateAccessHelper traverses ancestors but never checks n itself.
538
// For task validation we must also check whether n is paused.
541
return err
542
}
543
>
componentValue, _ := n.value.(Component) //nolint:revive // unchecked-type-assertion
tree.go ×2
544
>
if componentValue.LifecycleState(ctx).IsPaused() {
546
>
}
547
}
548
550
}
551
552
// validateAccessHelper is a helper method that validates both the current
553
// node's lifecycle state AND its ancestors recursively.
554
// Do not call this method directly, call validateAccess instead.
556
>
// Check ancestors first (if not detached).
557
>
if !n.isDetached() && n.parent != nil {
560
>
}
561
}
562
563
// Only Component nodes need to be validated.
566
>
}
567
568
// Hydrate the component so we can access its LifecycleState.
570
return err
571
}
572
>
componentValue, _ := n.value.(Component) //nolint:revive // unchecked-type-assertion
tree.go ×5
573
>
574
>
lifecycleState := componentValue.LifecycleState(ctx)
575
>
if lifecycleState.IsClosed() {
577
>
}
578
581
>
}
582
585
>
// This handles the case where root is terminated in the current transaction.
586
>
return errAccessCheckFailed
587
>
}
588
589
// terminated field check above is in memory only, so handle the case where root is terminated (closed)
590
// in a previous transaction and we have a mutable state reload which clears the field.
591
>
if n.parent == nil && n.backend.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
tree.go ×1
593
>
}
594
596
}
597
598
func (n *Node) prepareComponentValue(
599
chasmContext Context,
601
>
if n.valueState == valueStateNeedDeserialize {
603
>
componentAttr := metadata.GetComponentAttributes()
604
>
if componentAttr == nil {
605
return softassert.UnexpectedInternalErr(
606
n.logger,
607
"expect chasm node to have ComponentAttributes",
608
fmt.Errorf("actual attributes: %v", metadata.Attributes))
609
}
610
612
>
if !ok {
613
return softassert.UnexpectedInternalErr(
614
n.logger,
615
"unknown component type ID",
616
fmt.Errorf("%d", componentAttr.GetTypeId()))
617
}
618
620
return fmt.Errorf("failed to deserialize component: %w", err)
621
}
622
}
623
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.
627
>
if componentCanBeMutated {
629
>
}
630
632
}
633
634
func (n *Node) prepareDataValue(
635
chasmContext Context,
636
valueT reflect.Type,
638
>
metadata := n.serializedNode.Metadata
639
>
dataAttr := metadata.GetDataAttributes()
640
>
if dataAttr == nil {
641
return softassert.UnexpectedInternalErr(
642
n.logger,
643
"expect chasm node to have DataAttributes",
644
fmt.Errorf("actual attributes: %v", metadata.Attributes))
645
}
646
649
return fmt.Errorf("failed to deserialize data: %w", err)
650
}
651
}
652
653
// For now, we assume if a node is accessed with a MutableContext,
654
// its value will be mutated and no longer in sync with the serializedNode.
656
>
if componentCanBeMutated {
657
>
n.setValueState(valueStateNeedSerialize)
658
>
}
659
661
}
662
664
>
metadata := n.serializedNode.Metadata
665
>
pointerAttr := metadata.GetPointerAttributes()
666
>
if pointerAttr == nil {
667
return softassert.UnexpectedInternalErr(
668
n.logger,
669
"expect chasm node to have PointerAttributes",
670
fmt.Errorf("actual attributes: %v", metadata.Attributes))
671
}
672
674
if err := n.deserialize(nil); err != nil {
675
return fmt.Errorf("failed to deserialize data: %w", err)
676
}
677
}
678
680
}
681
683
>
return n.serializedNode.GetMetadata().GetComponentAttributes() != nil
684
>
}
685
687
>
return n.serializedNode.GetMetadata().GetDataAttributes() != nil
688
>
}
689
691
>
return n.serializedNode.GetMetadata().GetCollectionAttributes() != nil
692
>
}
693
695
>
componentAttr := n.serializedNode.GetMetadata().GetComponentAttributes()
696
>
if componentAttr == nil {
698
>
}
700
>
if componentTypeID == CallbackComponentID ||
701
>
componentTypeID == visibilityComponentTypeID {
702
>
// For backward compatibility purpose, we need to special handle callback and visibility components,
chasm_visibility.pb.go ×1
703
>
// which are implemented before detached component is properly supported by the framework.
704
>
return true
705
>
}
707
}
708
710
>
if n.serializedNode.GetMetadata().GetComponentAttributes() != nil {
712
>
}
713
716
>
}
717
719
>
return fieldTypePointer
720
>
}
721
722
if n.serializedNode.GetMetadata().GetCollectionAttributes() != nil {
723
softassert.Fail(
724
n.logger,
725
"fieldType can't be called on Collection node because Collection is not a Field")
726
}
727
728
return fieldTypeUnspecified
729
}
730
732
>
return fieldsOf(reflect.ValueOf(n.value))
733
>
}
734
736
>
if t == nil {
737
return nil
738
}
739
741
return serviceerror.NewInternalf("only pointer to struct is supported for tree node value: got %s", t.String())
742
}
744
}
745
747
>
switch ft {
749
>
n.serializedNode = &persistencespb.ChasmNode{
750
>
Metadata: &persistencespb.ChasmNodeMetadata{
751
>
InitialVersionedTransition: &persistencespb.VersionedTransition{
752
>
TransitionCount: n.backend.NextTransitionCount(),
753
>
NamespaceFailoverVersion: n.backend.GetCurrentVersion(),
754
>
},
755
>
Attributes: &persistencespb.ChasmNodeMetadata_DataAttributes{
756
>
DataAttributes: &persistencespb.ChasmDataAttributes{},
757
>
},
758
>
},
759
>
}
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
>
}
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,
787
"initSerializedNode can't be called with unspecified field type")
788
}
789
}
790
792
>
n.serializedNode = &persistencespb.ChasmNode{
793
>
Metadata: &persistencespb.ChasmNodeMetadata{
794
>
InitialVersionedTransition: &persistencespb.VersionedTransition{
795
>
TransitionCount: n.backend.NextTransitionCount(),
796
>
NamespaceFailoverVersion: n.backend.GetCurrentVersion(),
797
>
},
798
>
Attributes: &persistencespb.ChasmNodeMetadata_CollectionAttributes{
799
>
CollectionAttributes: &persistencespb.ChasmCollectionAttributes{},
800
>
},
801
>
},
802
>
}
803
>
}
804
805
func (n *Node) setSerializedNode(
806
nodePath []string,
807
encodedPath string,
808
serializedNode *persistencespb.ChasmNode,
810
>
if len(nodePath) == 0 {
811
>
n.serializedNode = serializedNode
812
>
n.setValueState(valueStateNeedDeserialize)
813
>
n.encodedPath = &encodedPath
814
>
return n
815
>
}
816
818
>
childNode, ok := n.children[childName]
819
>
if !ok {
820
>
childNode = newNode(n.nodeBase, n, childName)
821
>
n.children[childName] = childNode
822
>
}
823
>
return childNode.setSerializedNode(nodePath[1:], encodedPath, serializedNode)
824
}
825
826
// hasNewTransactionSideEffects returns true when the transaction has observable
827
// effects that must be persisted regardless of whether data bytes changed:
828
// new tasks scheduled on this node, or lifecycle termination.
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.
836
>
switch n.serializedNode.GetMetadata().GetAttributes().(type) {
838
>
return n.serializeComponentNode()
840
>
return n.serializeDataNode()
842
>
return n.serializeCollectionNode()
844
>
return n.serializePointerNode()
845
default:
846
return softassert.UnexpectedInternalErr(n.logger, "unknown node type", nil)
847
}
848
}
849
850
// serializeComponentNode serializes the component node.
851
// If this method is updated to modify serialized fields beyond Data and
852
// LastUpdateVersionedTransition, the skip-if-clean revert logic in
853
// closeTransactionSerializeNodes must be updated accordingly.
855
>
for field := range n.valueFields() {
856
>
if field.err != nil {
857
return field.err
858
}
859
862
}
863
865
>
if !field.val.IsNil() {
867
>
if blob, err = encodeChasmBlob(field.val.Interface().(proto.Message)); err != nil {
868
return err
869
}
870
}
871
873
>
874
>
if n.serializedNode.GetMetadata().GetLastUpdateVersionedTransition() == nil {
876
>
if !ok {
877
return softassert.UnexpectedInternalErr(
878
n.logger,
879
"component type is not registered",
880
fmt.Errorf("%s", reflect.TypeOf(n.value).String()))
881
}
882
// TypeId mismatch on a brand new node indicates node reassignment.
883
>
existingTypeID := n.serializedNode.GetMetadata().GetComponentAttributes().GetTypeId()
tree.go ×3
884
>
if existingTypeID != 0 && existingTypeID != rc.componentID {
885
return softassert.UnexpectedInternalErr(
886
n.logger,
887
"component node TypeId changed on first serialization",
888
fmt.Errorf("existing: %d, new: %d", existingTypeID, rc.componentID),
889
)
890
}
892
}
893
895
>
n.setValueState(valueStateSynced)
896
897
// continue to iterate over fields to validate that there is only one proto field in the component.
898
}
900
}
901
902
// syncSubComponents syncs the entire tree recursively (starting from the root node n) from the underlining component value:
903
// - Create:
904
// -- if child node is nil but subcomponent is not empty or key present in the collection,
905
// a new node with subcomponent/collection_item value is created.
906
// - Delete:
907
// -- if subcomponent is empty, the corresponding child is removed from the tree,
908
// -- if subcomponent is no longer in a component, the corresponding child is removed from the tree,
909
// -- if collection item is not in the collection, the corresponding child is removed from the tree,
910
// -- when a child is removed, all its children are removed too.
911
//
912
// All removed paths are added to mutation.DeletedNodes (which is shared between all nodes in the tree).
913
//
914
// True is returned when CHASM must perform deferred pointer resolution.
915
//
916
// nolint:revive,cognitive-complexity
918
>
if n.valueState < valueStateNeedSyncStructure {
921
>
if err != nil {
922
return err
923
}
924
}
926
}
927
929
>
for field := range n.valueFields() {
930
>
if field.err != nil {
931
return field.err
932
}
933
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")
939
// Nothing to sync.
941
>
keepChild, updatedFieldV, err := n.syncSubField(field.val, field.name)
942
>
if err != nil {
943
return err
944
}
947
>
}
950
>
}
952
>
internalField := field.val.FieldByName(parentPtrInternalFieldName)
953
>
internal, ok := internalField.Interface().(parentPtrInternal)
954
>
if !ok {
955
return softassert.UnexpectedInternalErr(
956
n.logger,
957
"CHASM parent pointer's internal field is not of parentPtrInternal type",
958
fmt.Errorf("node %s, actual type: %T", n.nodeName, internalField.Interface()))
959
}
962
>
internalField.Set(reflect.ValueOf(internal))
963
>
}
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,
969
"CHASM map must be of map type",
970
fmt.Errorf("node %s", n.nodeName))
971
}
972
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
980
>
if collectionNode == nil {
981
>
collectionNode = newNode(n.nodeBase, n, field.name)
982
>
collectionNode.initSerializedCollectionNode()
983
>
collectionNode.setValueState(valueStateNeedSyncStructure)
984
>
n.children[field.name] = collectionNode
985
>
}
986
988
>
if mapValT.Kind() != reflect.Struct || genericTypePrefix(mapValT) != chasmFieldTypePrefix {
989
return softassert.UnexpectedInternalErr(
990
n.logger,
991
"CHASM map value must be of Field[T] type",
992
fmt.Errorf("node %s got %s", n.nodeName, mapValT))
993
}
994
996
>
for _, mapKeyV := range field.val.MapKeys() {
997
>
mapItemV := field.val.MapIndex(mapKeyV)
998
>
collectionKey, err := n.mapKeyToString(mapKeyV)
999
>
if err != nil {
1000
return err
1001
}
1002
>
keepItem, updatedMapItemV, err := collectionNode.syncSubField(mapItemV, collectionKey)
tree.go ×9
1003
>
if err != nil {
1004
return err
1005
}
1007
>
// The only way to update item in the map is to set it back.
1008
>
field.val.SetMapIndex(mapKeyV, updatedMapItemV)
1009
>
}
1010
>
if keepItem {
1011
>
collectionItemsToKeep[collectionKey] = struct{}{}
1012
>
}
1013
}
1015
return err
1016
}
1017
>
collectionNode.setValueState(min(valueStateNeedSerialize, collectionNode.valueState))
tree.go ×9
1018
>
childrenToKeep[field.name] = struct{}{}
1019
}
1020
}
1021
1023
>
n.setValueState(valueStateNeedSerialize)
1024
>
1025
>
return err
1026
}
1027
1029
>
switch keyV.Kind() {
1030
>
case reflect.String:
1031
>
return keyV.String(), nil
1032
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
1033
return strconv.FormatInt(keyV.Int(), 10), nil
1034
case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
1035
return strconv.FormatUint(keyV.Uint(), 10), nil
1036
case reflect.Bool:
1037
return strconv.FormatBool(keyV.Bool()), nil
1038
default:
1039
return "", softassert.UnexpectedInternalErr(
1040
n.logger,
1041
"CHASM map key type is not supported",
1042
fmt.Errorf("node %s must be one of [%s], got %s", n.nodeName, mapKeyTypes, keyV.Type().String()))
1043
}
1044
}
1045
1046
>
func (n *Node) stringToMapKey(nodeName string, key string, keyT reflect.Type) (reflect.Value, error) {
tree.go ×7
1047
>
var (
1048
>
keyV reflect.Value
1049
>
err error
1050
>
)
1051
>
switch keyT.Kind() {
1052
>
case reflect.String:
1053
>
keyV = reflect.ValueOf(key)
1054
case reflect.Int:
1055
var x int64
1056
x, err = strconv.ParseInt(key, 10, 0)
1057
keyV = reflect.ValueOf(int(x))
1058
case reflect.Int8:
1059
var x int64
1060
x, err = strconv.ParseInt(key, 10, 8)
1061
keyV = reflect.ValueOf(int8(x))
1062
case reflect.Int16:
1063
var x int64
1064
x, err = strconv.ParseInt(key, 10, 16)
1065
keyV = reflect.ValueOf(int16(x))
1066
case reflect.Int32:
1067
var x int64
1068
x, err = strconv.ParseInt(key, 10, 32)
1069
keyV = reflect.ValueOf(int32(x))
1070
case reflect.Int64:
1071
var x int64
1072
x, err = strconv.ParseInt(key, 10, 64)
1073
keyV = reflect.ValueOf(x)
1074
case reflect.Uint:
1075
var x uint64
1076
x, err = strconv.ParseUint(key, 10, 0)
1077
keyV = reflect.ValueOf(uint(x))
1078
case reflect.Uint8:
1079
var x uint64
1080
x, err = strconv.ParseUint(key, 10, 8)
1081
keyV = reflect.ValueOf(uint8(x))
1082
case reflect.Uint16:
1083
var x uint64
1084
x, err = strconv.ParseUint(key, 10, 16)
1085
keyV = reflect.ValueOf(uint16(x))
1086
case reflect.Uint32:
1087
var x uint64
1088
x, err = strconv.ParseUint(key, 10, 32)
1089
keyV = reflect.ValueOf(uint32(x))
1090
case reflect.Uint64:
1091
var x uint64
1092
x, err = strconv.ParseUint(key, 10, 64)
1093
keyV = reflect.ValueOf(x)
1094
case reflect.Bool:
1095
var b bool
1096
b, err = strconv.ParseBool(key)
1097
keyV = reflect.ValueOf(b)
1098
default:
1099
// Use softassert only here because this is the only case that indicates "compile" time error.
1100
// The other errors below can come from data type mismatch between a component and persisted data.
1101
err = softassert.UnexpectedInternalErr(
1102
n.logger,
1103
"unsupported CHASM map key type",
1104
fmt.Errorf("unsupported type %s of kind %s: supported key types: %s", keyT.String(), keyT.Kind().String(), mapKeyTypes),
1105
tag.Error(err))
1106
}
1107
1109
err = fmt.Errorf("value %s is not valid of type %s of kind %s", key, keyT.String(), keyT.Kind().String())
1110
}
1111
1113
err = softassert.UnexpectedInternalErr(
1114
n.logger,
1115
"serialized map key value can't be parsed to CHASM map key type",
1116
fmt.Errorf("nodeName: %s, key: %s, keyType: %s, error: %s", nodeName, key, keyT.String(), err.Error()))
1117
}
1118
1120
}
1121
1122
// syncSubField syncs node n with value from fieldV parameter.
1123
// If fieldV is a component, then it will sync all subcomponents recursively.
1124
// It returns:
1125
// - bool keepNode indicates if node needs to be removed from parent's children map.
1126
// - updatedFieldV if fieldV needs to be updated with new value.
1127
// If updatedFieldV is invalid, then fieldV doesn't need to be updated.
1128
// NOTE: this function doesn't update fieldV because it might come from the map which is not addressable.
1129
// - error.
1130
func (n *Node) syncSubField(
1131
fieldV reflect.Value,
1132
fieldN string,
1133
) (
1134
keepNode bool,
1135
updatedFieldV reflect.Value,
1136
err error,
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 ×1
1143
>
// In this case, return keepNode=false and this node (and all it children) will be added to DeletedNodes map.
1144
>
return
1145
>
}
1146
1148
>
if internal.node == nil && fieldValue != nil {
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 {
1159
>
if err = assertStructPointer(reflect.TypeOf(fieldValue)); err != nil {
1160
return
1161
}
1162
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
}
1174
>
if err = assertStructPointer(reflect.TypeOf(fieldValue)); err != nil {
1175
return
1176
}
1177
case fieldTypePointer:
1178
if _, ok := fieldValue.([]string); !ok {
1179
err = softassert.UnexpectedInternalErr(
1180
n.logger,
1181
"value must be of type []string for the field of pointer type",
1182
fmt.Errorf("got %T", fieldValue))
1183
return
1184
}
1186
>
n.needsPointerResolution = true
1187
default:
1188
err = softassert.UnexpectedInternalErr(
1189
n.logger,
1190
"unexpected field type",
1191
fmt.Errorf("%d", fieldType),
1192
)
1193
return
1194
}
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
1206
>
if err != nil {
1207
return
1208
}
1209
}
1210
1212
}
1213
1214
func (n *Node) deleteChildren(
1215
childrenToKeep map[string]struct{},
1217
>
for childName, childNode := range n.children {
1220
return err
1221
}
1222
}
1223
}
1225
}
1226
1227
// serializeDataNode serializes the data node.
1228
// If this method is updated to modify serialized fields beyond Data and
1229
// LastUpdateVersionedTransition, the skip-if-clean revert logic in
1230
// closeTransactionSerializeNodes must be updated accordingly.
1232
>
protoValue, ok := n.value.(proto.Message)
1233
>
if !ok {
1234
return serviceerror.NewInternal("only support proto.Message as chasm data")
1235
}
1236
1238
>
if protoValue != nil {
1239
>
var err error
1240
>
if blob, err = encodeChasmBlob(protoValue); err != nil {
1241
return err
1242
}
1243
}
1245
>
n.updateLastUpdateVersionedTransition()
1246
>
n.setValueState(valueStateSynced)
1247
>
1248
>
return nil
1249
}
1250
1251
// serializeCollectionNode serializes the collection node.
1252
// If this method is updated to modify serialized fields beyond
1253
// LastUpdateVersionedTransition, the skip-if-clean revert logic in
1254
// closeTransactionSerializeNodes must be updated accordingly.
1256
>
// The collection node has no data; therefore, only metadata needs to be updated.
1257
>
n.updateLastUpdateVersionedTransition()
1258
>
n.setValueState(valueStateSynced)
1259
>
return nil
1260
>
}
1261
1262
// serializePointerNode doesn't serialize anything but named this way for consistency.
1264
>
path, isPathValid := n.value.([]string)
1265
>
if !isPathValid {
1266
return softassert.UnexpectedInternalErr(
1267
n.logger,
1268
"pointer path is not []string",
1269
fmt.Errorf("got %T for node %s", n.value, n.nodeName))
1270
}
1271
1273
>
n.updateLastUpdateVersionedTransition()
1274
>
n.setValueState(valueStateSynced)
1275
>
1276
>
return nil
1277
}
1278
1280
>
if n.serializedNode.GetMetadata().GetLastUpdateVersionedTransition() == nil {
1281
>
n.serializedNode.GetMetadata().LastUpdateVersionedTransition = &persistencespb.VersionedTransition{}
tree.go ×1
1282
>
}
1283
>
n.serializedNode.GetMetadata().GetLastUpdateVersionedTransition().TransitionCount = n.backend.NextTransitionCount()
tree.go ×2
1284
>
n.serializedNode.GetMetadata().GetLastUpdateVersionedTransition().NamespaceFailoverVersion = n.backend.GetCurrentVersion()
1285
}
1286
1287
// deserialize initializes the node's value from its serializedNode.
1288
// If a value is of the component type, it initializes every chasm.Field of it and sets serializedNode field but not value field,
1289
// i.e., it doesn't deserialize recursively and must be called on every node separately.
1290
// valueT must be a pointer to a concrete type (not interface). To support deserialization of a component to interface,
1291
// a registry lookup must be done outside the deserialize method.
1292
func (n *Node) deserialize(
1293
valueT reflect.Type,
1295
>
if err := assertStructPointer(valueT); err != nil {
1296
return err
1297
}
1298
1299
>
if n.valueState != valueStateNeedDeserialize && reflect.TypeOf(n.value) == valueT {
tree.go ×8
1301
>
}
1302
1304
>
case *persistencespb.ChasmNodeMetadata_ComponentAttributes:
1305
>
return n.deserializeComponentNode(valueT)
1307
>
return n.deserializeDataNode(valueT)
1308
case *persistencespb.ChasmNodeMetadata_CollectionAttributes:
1309
softassert.Fail(
1310
n.logger,
1311
"deserialize shouldn't be called on the collection node because it is deserialized with the parent component.")
1312
case *persistencespb.ChasmNodeMetadata_PointerAttributes:
1313
return n.deserializePointerNode()
1314
}
1315
return nil
1316
}
1317
1318
func (n *Node) deserializeComponentNode(
1319
valueT reflect.Type,
1321
>
// valueT is guaranteed to be a pointer to the struct because it was already validated by the assertStructPointer method.
1322
>
valueV := reflect.New(valueT.Elem())
1323
>
1324
>
for field := range fieldsOf(valueV) {
1325
>
if field.err != nil {
1326
return field.err
1327
}
1328
1330
case fieldKindUnspecified:
1331
softassert.Fail(
1332
n.logger,
1333
"field.kind can be unspecified only if err is not nil, and there is a check for it above",
1334
tag.String("node name", n.nodeName))
1336
>
value, err := unmarshalProto(n.serializedNode.GetData(), field.typ)
1337
>
if err != nil {
1338
return err
1339
}
1342
>
if childNode, found := n.children[field.name]; found {
1344
>
internalValue := reflect.ValueOf(newFieldInternalWithNode(childNode))
1345
>
chasmFieldV.FieldByName(internalFieldName).Set(internalValue)
1346
>
field.val.Set(chasmFieldV)
1347
>
}
1349
>
if collectionNode, found := n.children[field.name]; found {
1351
>
if mapFieldV.IsNil() {
1352
>
mapFieldV = reflect.MakeMapWithSize(field.typ, field.val.Len())
1353
>
field.val.Set(mapFieldV)
1354
>
}
1355
1357
>
// field.typ.Elem() is a go type of map item: Field[T]
1358
>
chasmFieldV := reflect.New(field.typ.Elem()).Elem()
1359
>
internalValue := reflect.ValueOf(newFieldInternalWithNode(collectionItemNode))
1360
>
chasmFieldV.FieldByName(internalFieldName).Set(internalValue)
1361
>
mapKeyV, err := n.stringToMapKey(field.name, collectionItemName, mapFieldV.Type().Key())
1362
>
if err != nil {
1363
return err
1364
}
1366
}
1368
>
field.val.Set(reflect.MakeMap(field.typ))
1369
>
}
1370
>
case fieldKindMutableState:
1371
>
field.val.Set(reflect.ValueOf(NewMSPointer(n.backend)))
1373
>
parentPtrV := reflect.New(field.typ).Elem()
1374
>
parentPtrV.FieldByName(parentPtrInternalFieldName).Set(reflect.ValueOf(parentPtrInternal{
1375
>
currentNode: n,
1376
>
}))
1377
>
field.val.Set(parentPtrV)
1378
}
1379
}
1380
1382
>
n.setValueState(valueStateSynced)
1383
>
return nil
1384
}
1385
1386
func (n *Node) deserializeDataNode(
1387
valueT reflect.Type,
1389
>
value, err := unmarshalProto(n.serializedNode.GetData(), valueT)
1390
>
if err != nil {
1391
return err
1392
}
1393
1395
>
n.setValueState(valueStateSynced)
1396
>
return nil
1397
}
1398
1399
// deserializePointerNode doesn't deserialize anything but named this way for consistency.
1400
func (n *Node) deserializePointerNode() error {
1401
n.setValue(n.serializedNode.GetMetadata().GetPointerAttributes().GetNodePath())
1402
n.setValueState(valueStateSynced)
1403
return nil
1404
}
1405
1406
func unmarshalProto(
1407
dataBlob *commonpb.DataBlob,
1408
valueT reflect.Type,
1410
>
if !valueT.AssignableTo(protoMessageT) {
1411
return reflect.Value{}, serviceerror.NewInternal("only support proto.Message as chasm data")
1412
}
1413
1415
>
1416
>
if dataBlob == nil || len(dataBlob.Data) == 0 {
1417
>
// If the original data is the zero value of its type, the dataBlob loaded from persistence layer will be nil.
tree.go ×1
1418
>
// But we know for component & data nodes, they won't get persisted in the first place if there's no data,
1419
>
// so it must be a zero value.
1420
>
dataBlob = &commonpb.DataBlob{
1421
>
EncodingType: enumspb.ENCODING_TYPE_PROTO3,
1422
>
Data: []byte{},
1423
>
}
1424
>
}
1425
1426
>
if err := serialization.Decode(dataBlob, value.Interface().(proto.Message)); err != nil {
tree.go ×4
1427
return reflect.Value{}, err
1428
}
1429
1431
}
1432
1433
// Ref implements the CHASM Context interface
1434
func (n *Node) Ref(
1435
component Component,
1437
>
ref, err := n.structuredRef(component)
1438
>
if err != nil {
1440
>
}
1442
}
1443
1444
// structuredRef returns a ComponentRef for the node.
1445
func (n *Node) structuredRef(
1446
component Component,
1448
>
// No need to update tree structure here. If a Component can only be found after
1449
>
// syncSubComponents() is called, it means the component is created in the
1450
>
// current transition and don't have a reference yet.
1451
>
1452
>
refNode, ok := n.valueToNode[component]
1453
>
if !ok || !refNode.isComponent() {
1455
>
}
1456
1458
>
return ComponentRef{
1459
>
ExecutionKey: ExecutionKey{
1460
>
NamespaceID: workflowKey.NamespaceID,
1461
>
BusinessID: workflowKey.WorkflowID,
1462
>
RunID: workflowKey.RunID,
1463
>
},
1464
>
archetypeID: n.ArchetypeID(),
1465
>
// TODO: Consider using node's LastUpdateVersionedTransition for checking staleness here.
1466
>
// Using VersionedTransition of the entire tree might be too strict.
1467
>
executionLastUpdateVT: transitionhistory.CopyVersionedTransition(refNode.backend.CurrentVersionedTransition()),
1468
>
componentPath: refNode.path(),
1469
>
componentInitialVT: refNode.serializedNode.GetMetadata().GetInitialVersionedTransition(),
1470
>
}, nil
1471
1472
}
1473
1474
// componentLinks returns the union of links across all requests stored on the
1475
// given component's metadata. Pending writes staged in the current transaction
1476
// replace persisted entries for the same request ID (matching the read
1477
// semantics of componentRequestLinks), so a caller staging an update doesn't
1478
// observe stale + new entries side-by-side.
1480
>
var links []*commonpb.Link
1481
>
pending := n.pendingRequestLinks[component]
1482
>
1483
>
for _, ls := range pending {
1485
>
}
1486
1488
>
for requestID, req := range refNode.serializedNode.GetMetadata().GetComponentAttributes().GetRequests() {
1489
>
if _, overridden := pending[requestID]; overridden {
1491
}
1493
}
1494
}
1495
1497
}
1498
1499
// setComponentRequestLinks records the links contributed by the given request
1500
// on the component, replacing any prior entry for the same request ID. Passing
1501
// nil/empty links removes the entry. The write is staged and applied during
1502
// CloseTransaction, so it works for components that have not yet been
1503
// registered as nodes. An empty requestID is rejected to avoid silent
1504
// collisions across callers.
1505
>
func (n *Node) setComponentRequestLinks(component Component, requestID string, links []*commonpb.Link) error {
context.go ×1
1506
>
if requestID == "" {
1507
>
return serviceerror.NewInvalidArgument("requestID is required when setting per-request links")
tree.go ×2
1508
>
}
1510
>
if !ok {
1511
>
perRequest = make(map[string][]*commonpb.Link)
1512
>
n.pendingRequestLinks[component] = perRequest
1513
>
}
1514
>
perRequest[requestID] = links
1515
>
return nil
1516
}
1517
1518
// componentRequestLinks returns the links stored on the given component's
1519
// metadata for the specific requestID, preferring a pending write staged in
1520
// the current transaction. Returns nil if no entry exists.
1521
>
func (n *Node) componentRequestLinks(component Component, requestID string) ([]*commonpb.Link, error) {
context.go ×1
1522
>
if requestID == "" {
1523
>
return nil, serviceerror.NewInvalidArgument("requestID is required when reading per-request links")
tree.go ×2
1524
>
}
1527
>
return links, nil
1528
>
}
1529
}
1531
>
if req, ok := refNode.serializedNode.GetMetadata().GetComponentAttributes().GetRequests()[requestID]; ok {
1532
>
return req.GetLinks(), nil
1533
>
}
1534
}
1535
return nil, nil
1536
}
1537
1538
// componentUserMetadata returns the user metadata stored on the given
1539
// component, preferring a pending write staged in the current transaction.
1541
>
if md, ok := n.pendingUserMetadata[component]; ok {
1542
return md
1543
}
1545
>
return refNode.serializedNode.GetMetadata().GetComponentAttributes().GetUserMetadata()
1546
>
}
1547
return nil
1548
}
1549
1550
// setComponentUserMetadata stages a user-metadata write for the given component.
1551
// Applied during CloseTransaction.
1552
>
func (n *Node) setComponentUserMetadata(component Component, md *sdkpb.UserMetadata) error {
context.go ×1
1553
>
n.pendingUserMetadata[component] = md
1554
>
return nil
1555
>
}
1556
1557
// closeTransactionApplyPendingComponentMetadata walks the tree and applies any
1558
// staged framework metadata (request links, user metadata) to each component
1559
// node, marking touched nodes as updated for replication. Pending entries that
1560
// reference a component that was never registered as a node are dropped and
1561
// logged at warn level to surface caller misuse.
1563
>
if len(n.pendingRequestLinks) == 0 && len(n.pendingUserMetadata) == 0 {
1564
>
return nil
1565
>
}
1567
>
if !node.applyPendingComponentMetadata() {
1568
>
continue
1569
}
1571
>
if err != nil {
1572
return err
1573
}
1575
>
node.updateLastUpdateVersionedTransition()
1576
>
n.mutation.UpdatedNodes[encodedPath] = node.serializedNode
1577
>
delete(n.mutation.DeletedNodes, encodedPath)
1578
>
}
1579
}
1582
>
"chasm: dropped staged component metadata for components that were never registered as nodes",
1583
>
tag.NewInt("orphan-request-link-components", len(n.pendingRequestLinks)),
1584
>
tag.NewInt("orphan-user-metadata-components", len(n.pendingUserMetadata)),
1585
>
)
1586
>
}
1588
>
n.pendingUserMetadata = make(map[any]*sdkpb.UserMetadata)
1589
>
return nil
1590
}
1591
1592
// applyPendingComponentMetadata writes staged per-component framework metadata
1593
// (request links and user metadata) onto the node's ChasmComponentAttributes.
1594
// Returns true if the node was mutated.
1596
>
if n.value == nil {
1597
return false
1598
}
1600
>
if attrs == nil {
1601
>
return false
1602
>
}
1603
>
dirty := false
1604
>
1605
>
if pending, ok := n.pendingRequestLinks[n.value]; ok {
1607
>
attrs.Requests = make(map[string]*persistencespb.ChasmComponentAttributes_RequestMetadata)
1608
>
}
1609
>
for requestID, links := range pending {
1610
>
if len(links) == 0 {
1612
>
delete(attrs.Requests, requestID)
1613
>
dirty = true
1614
>
}
1615
>
continue
1616
}
1617
>
attrs.Requests[requestID] = &persistencespb.ChasmComponentAttributes_RequestMetadata{Links: links}
tree.go ×3
1618
>
dirty = true
1619
}
1621
}
1622
1625
>
dirty = true
1626
>
delete(n.pendingUserMetadata, n.value)
1627
>
}
1628
1630
}
1631
1632
// componentNodePath implements the CHASM Context interface
1633
func (n *Node) componentNodePath(
1634
component Component,
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() {
1642
>
}
1643
1645
}
1646
1647
// dataNodePath implements the CHASM Context interface
1648
func (n *Node) dataNodePath(
1649
data proto.Message,
1650
) ([]string, error) {
1651
// It's unnecessary to deserialize entire tree as calling this method means
1652
// caller already have the deserialized value.
1653
1654
refNode, ok := n.valueToNode[data]
1655
if !ok || !refNode.isData() {
1656
return nil, errDataNotFound
1657
}
1658
1659
return refNode.path(), nil
1660
}
1661
1662
// Now implements the CHASM Context interface
1663
func (n *Node) Now(
1664
_ Component,
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
1671
func (n *Node) AddTask(
1672
component Component,
1673
taskAttributes TaskAttributes,
1674
task any,
1676
>
rt, ok := n.registry.taskFor(task)
1677
>
if ok && rt.isPureTask && taskAttributes.IsImmediate() {
1679
>
n.immediatePureTasks[component] = append(n.immediatePureTasks[component], taskWithAttributes{
1680
>
task: task,
1681
>
attributes: taskAttributes,
1682
>
})
1683
>
return
1684
>
}
1685
1687
>
task: task,
1688
>
attributes: taskAttributes,
1689
>
})
1690
}
1691
1692
// CloseTransaction is used by MutableState to close the transaction and
1693
// track changes made in the current transaction.
1695
>
defer n.cleanupTransaction()
1696
>
1697
>
if err := n.executeImmediatePureTasks(); err != nil {
1698
return NodesMutation{}, err
1699
}
1700
1702
return NodesMutation{}, err
1703
}
1704
1708
>
}
1709
}
1710
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
1723
>
if err := n.closeTransactionForceUpdateVisibility(immutableContext, rootLifecycleChanged); err != nil {
tree.go ×9
1724
return NodesMutation{}, err
1725
}
1726
}
1727
1729
return NodesMutation{}, err
1730
}
1731
1732
>
if err := n.closeTransactionUpdateComponentTasks(nextVersionedTransition); err != nil {
tree.go ×18
1733
return NodesMutation{}, err
1734
}
1735
1737
return NodesMutation{}, err
1738
}
1739
1740
// Both user & system data mutation need to be returned and persisted.
1742
>
maps.Copy(n.mutation.DeletedNodes, n.systemMutation.DeletedNodes)
1743
>
1744
>
return n.mutation, nil
1745
}
1746
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 {
1757
>
// added while existing ones are executed.
1758
>
immediatePureTasks := n.immediatePureTasks
1759
>
n.immediatePureTasks = make(map[any][]taskWithAttributes)
1760
>
1761
>
for component, pureTasks := range immediatePureTasks {
1762
>
for _, task := range pureTasks {
1763
>
if syncStructure {
1764
>
if err := n.syncSubComponents(); err != nil {
1765
return err
1766
}
1767
}
1768
1769
// The corresponding Node may not be found due to several reasons:
1770
// 1. This function is executed at the end of a transaction which could contain multiple transitions.
1771
// So it's possible that a task added for a component in one transition and in a later transition that component get removed.
1772
// 2. Previous pure task for the node deleted the node itself via a (parent) pointer.
1773
// This is also why this check is done in the inner for loop.
1775
>
if !ok {
1777
}
1778
1779
// Only syncStructure on next iteration if task is executed (the first return value).
1780
>
syncStructure, err = taskNode.ExecutePureTask(context.Background(), task.attributes, task.task)
tree.go ×4
1781
>
if err != nil {
1782
return err
1783
}
1784
}
1785
}
1786
}
1787
1789
}
1790
1791
func (n *Node) closeTransactionHandleRootLifecycleChange(
1792
immutableContext Context,
1794
>
if n.backend.IsWorkflow() {
1796
>
return false, nil
1797
>
}
1798
1801
>
}
1802
1803
>
if n.backend.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
tree.go ×2
1805
>
return false, nil
1806
>
}
1807
1810
>
enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
1811
>
enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED,
1812
>
)
1813
>
}
1814
1816
>
if err != nil {
1817
return false, err
1818
}
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
1829
>
newState = enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED
1830
>
newStatus = enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
1832
>
newState = enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED
1833
>
newStatus = enumspb.WORKFLOW_EXECUTION_STATUS_FAILED
1834
default:
1835
return false, softassert.UnexpectedInternalErr(
1836
n.logger,
1837
"unknown component lifecycle state",
1838
fmt.Errorf("%v", lifecycleState))
1839
}
1840
1842
}
1843
1844
func (n *Node) closeTransactionForceUpdateVisibility(
1845
immutableContext Context,
1846
rootLifecycleChanged bool,
1848
>
if n.deleteAfterClose {
1850
>
}
1851
1853
>
n.backend.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
1855
>
}
1856
1858
>
1859
>
rootComponent, err := n.Component(immutableContext, ComponentRef{})
1860
>
if err != nil {
1861
return err
1862
}
1863
1865
>
if ok {
1867
>
newSA := searchAttributeKeyValuesToMap(saSlice)
1868
>
if !maps.EqualFunc(n.currentSA, newSA, isVisibilityValueEqual) {
1870
>
}
1872
}
1873
1875
>
if ok {
1877
>
if !proto.Equal(n.currentMemo, newMemo) {
1879
>
}
1881
}
1882
1885
>
}
1886
1888
>
for _, child := range n.children {
1891
}
1892
1894
>
if rc, ok := n.registry.componentFor(child.value); ok && rc.fqType() == visibilityComponentType {
tree.go ×1
1896
>
break
1897
}
1898
>
} else if child.serializedNode.Metadata.GetComponentAttributes().TypeId == visibilityComponentTypeID {
tree.go ×1
1900
>
break
1901
}
1902
}
1903
1906
>
}
1907
1909
>
if err != nil {
1910
return err
1911
}
1912
1914
>
if !ok {
1915
return softassert.UnexpectedInternalErr(
1916
n.logger,
1917
"expected visibility component for component type",
1918
fmt.Errorf("type: %s, but got %T", visibilityComponentType, visComponent))
1919
}
1920
1921
// Generate a task and mark the node as dirty.
1922
//
1923
// NOTE: generateTask() will create a new logical task for the visibility component. But it also
1924
// invalidates all previous logical tasks at the end of the transaction, and only one physical task
1925
// will be created in the visibility queue.
1927
>
visibility.generateTask(mutableContext)
1928
>
visibilityNode.setValueState(valueStateNeedSerialize)
1929
>
1930
>
// We don't need to sync tree structure here for the visiblity node because we only generated a task without
1931
>
// changing any component fields.
1932
>
return nil
1933
}
1934
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
1943
}
1944
1946
>
if err != nil {
1947
return err
1948
}
1949
1950
// Skip writing nodes whose serialized content hasn't changed. A nil
1951
// LastUpdateVersionedTransition means the node is brand new and must be written.
1952
// prevData captures the pre-serialize blob pointer; serialize() allocates a new
1953
// blob, leaving prevData pointing at the original for comparison.
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 {
1963
>
}
1964
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 ×4
1971
>
node.serializedNode.GetMetadata().LastUpdateVersionedTransition = prevVersionedTransition
tree.go ×1
1972
>
continue
1973
}
1974
1975
>
if componentAttr := node.serializedNode.GetMetadata().GetComponentAttributes(); componentAttr != nil &&
tree.go ×2
1976
>
componentAttr.TypeId == visibilityComponentTypeID &&
1977
>
len(nodePath) != 1 {
1978
return softassert.UnexpectedInternalErr(
1979
n.logger,
1980
"CHASM visibility component must be immediate child of the root node",
1981
fmt.Errorf("found at path %s", nodePath))
1982
}
1983
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
1995
}
1996
1997
func (n *Node) closeTransactionUpdateComponentTasks(
1998
nextVersionedTransition *persistencespb.VersionedTransition,
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 {
2013
}
2014
2015
// First update component logical tasks.
2016
2017
// Even if a node is not touched in this transaction, its task can still become invalid due to, e.g.
2018
// - child component state update
2019
// - parent component closing (access rule)
2020
// - a pointer field pointing to an updated component (pointers are ancestors-only)
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.
2025
>
if err := node.prepareComponentValue(taskValidationContext); err != nil {
2026
return err
2027
}
2028
2030
>
if err != nil {
2031
return err
2032
}
2033
2036
>
encodedPath, err := node.getEncodedPath()
2037
>
if err != nil {
2038
return err
2039
}
2042
>
node.updateLastUpdateVersionedTransition()
2043
>
2044
>
n.mutation.UpdatedNodes[encodedPath] = node.serializedNode
2045
>
delete(n.mutation.DeletedNodes, encodedPath)
2046
>
}
2047
}
2048
}
2049
2050
// The conditions excludes replication logic (applyMutation/Snapshot) which sets
2051
// valueState to valueStateNeedDeserialize.
2052
//
2053
// Do NOT use condition node.valueState == valueStateNeedSerialize.
2054
// This method is called after the closeTransactionSerializeNodes which sets valueState
2055
// to valueStateSynced.
2057
>
node.serializedNode.GetMetadata().LastUpdateVersionedTransition,
2058
>
nextVersionedTransition,
2059
>
) == 0 && node.valueState != valueStateNeedDeserialize {
2061
>
nextVersionedTransition,
2062
>
taskValidationContext,
2063
>
&taskOffset,
2064
>
); err != nil {
2065
return err
2066
}
2067
}
2068
2070
>
for idx := len(sideEffectTasks) - 1; idx >= 0; idx-- {
2072
>
if sideEffectTask.PhysicalTaskStatus == physicalTaskStatusCreated {
2074
}
2075
2077
>
sideEffectTask,
2078
>
nodePath,
2079
>
archetypeID,
2080
>
)
2081
}
2082
2083
// Find the first pure task in the entire tree,
2084
// regardless if the pure task is newly added or existing.
2086
>
if len(pureTasks) == 0 {
2088
}
2089
2091
>
comparePureTasks(pureTasks[0], firstPureTask) < 0 {
2092
>
firstPureTask = pureTasks[0]
2093
>
firstPureTaskNode = node
2094
>
}
2095
}
2096
2097
// TODO: We cannot simply assert that all tasks in n.nodeBase.newTasks are processed.
2098
// That should be the case when only one transition for each transaction.
2099
// However, when processing pure tasks, we run multiple pure tasks, thus multiple transitions
2100
// in one transaction. This means it's possible that task generated for a component in the first
2101
// task, and that component get deleted by the second task.
2102
2104
>
firstPureTask,
2105
>
firstPureTaskNode,
2106
>
archetypeID,
2107
>
)
2108
}
2109
2110
func (n *Node) deserializeComponentTask(
2111
componentTask *persistencespb.ChasmComponentAttributes_Task,
2113
>
registableTask, ok := n.registry.TaskByID(componentTask.TypeId)
2114
>
if !ok {
2115
return nil, softassert.UnexpectedInternalErr(
2116
n.logger,
2117
"unknown task type id",
2118
fmt.Errorf("%d", componentTask.TypeId))
2119
}
2120
2122
>
if err != nil {
2123
return nil, err
2124
}
2125
2127
}
2128
2129
// validateTask runs taskInstance's registered validation handler.
2130
// This method assumes component value is already hydrated.
2131
func (n *Node) validateTask(
2132
validateContext Context,
2133
taskInvocation TaskInvocation,
2134
taskInstance any,
2136
>
registableTask, ok := n.registry.taskFor(taskInstance)
2137
>
if !ok {
2138
return false, softassert.UnexpectedInternalErr(
2139
n.logger,
2140
"task type for goType is not registered",
2141
fmt.Errorf("%s", reflect.TypeOf(taskInstance).Name()))
2142
}
2143
2144
// checkPaused=true: a single ancestor walk invalidates tasks for both
2145
// closed ancestors and paused components (self or non-detached ancestor).
2148
>
return false, nil
2149
>
}
2150
return false, err
2151
}
2152
2154
>
2155
>
return registableTask.validateFn(
2156
>
validateContext,
2157
>
n.value,
2158
>
taskInvocation,
2159
>
taskInstance,
2160
>
n.registry,
2161
>
)
2162
}
2163
2164
func (n *Node) closeTransactionCleanupInvalidTasks(
2165
validateContext Context,
2167
>
// Validate existing tasks and remove invalid ones.
2168
>
var validationErr error
2169
>
cleanedUp := false
2170
>
deleteFunc := func(existingTask *persistencespb.ChasmComponentAttributes_Task) bool {
2172
>
if err != nil {
2173
validationErr = err
2174
return false
2175
}
2176
2178
>
validateContext,
2179
>
TaskInvocation{
2180
>
TaskAttributes: TaskAttributes{
2181
>
ScheduledTime: existingTask.ScheduledTime.AsTime(),
2182
>
Destination: existingTask.Destination,
2183
>
},
2184
>
},
2185
>
existingTaskInstance,
2186
>
)
2187
>
if err != nil {
2188
validationErr = err
2189
return false
2190
}
2193
>
delete(n.taskValueCache, existingTask.Data)
2194
>
}
2196
}
2197
2199
>
componentAttr.SideEffectTasks = slices.DeleteFunc(componentAttr.SideEffectTasks, deleteFunc)
2200
>
if validationErr != nil {
2201
return false, validationErr
2202
}
2204
>
if validationErr != nil {
2205
return false, validationErr
2206
}
2208
}
2209
2210
// applySingletonMode enforces singleton semantics for tasks registered with [WithSingletonTask].
2211
// It is called after task validation, so only valid new tasks reach this point.
2212
// Returns true if the new task should be skipped (SingletonTaskModeIgnore with existing task).
2213
func (n *Node) applySingletonMode(
2214
rt *RegistrableTask,
2215
taskList *[]*persistencespb.ChasmComponentAttributes_Task,
2217
>
if rt.singletonMode == 0 {
2219
>
}
2220
2221
>
idx := slices.IndexFunc(*taskList, func(t *persistencespb.ChasmComponentAttributes_Task) bool {
tree.go ×2
2223
>
})
2225
>
return false
2226
>
}
2227
2230
>
delete(n.taskValueCache, (*taskList)[idx].Data)
2231
>
*taskList = slices.Delete(*taskList, idx, idx+1)
2232
>
return false
2234
>
return true
2235
default:
2236
return false
2237
}
2238
}
2239
2240
func (n *Node) closeTransactionHandleNewTasks(
2241
nextVersionedTransition *persistencespb.VersionedTransition,
2242
validateContext Context,
2243
taskOffset *int64,
2245
>
newTasks, ok := n.newTasks[n.value]
2246
>
if !ok {
2248
>
}
2249
2251
>
sortPureTasks := false
2252
>
2253
>
for _, newTask := range newTasks {
2254
>
if !newTask.attributes.IsValid() {
2255
return softassert.UnexpectedInternalErr(
2256
n.logger,
2257
"task attributes cannot have both destination and scheduled specified",
2258
fmt.Errorf("attributes: %v", newTask.attributes))
2259
}
2260
2262
>
validateContext,
2263
>
TaskInvocation{TaskAttributes: newTask.attributes},
2264
>
newTask.task,
2265
>
)
2266
>
if err != nil {
2267
return err
2268
}
2271
}
2272
2274
>
if !ok {
2275
return softassert.UnexpectedInternalErr(
2276
n.logger,
2277
"task type is not registered",
2278
fmt.Errorf("%s", reflect.TypeOf(newTask.task).String()))
2279
}
2280
2281
>
taskBlob, err := n.serializeTaskWithCache(registrableTask, reflect.ValueOf(newTask.task))
tree.go ×12
2282
>
if err != nil {
2283
return err
2284
}
2285
2287
>
TypeId: registrableTask.taskTypeID,
2288
>
Destination: newTask.attributes.Destination,
2289
>
ScheduledTime: timestamppb.New(newTask.attributes.ScheduledTime),
2290
>
Data: taskBlob,
2291
>
VersionedTransition: nextVersionedTransition,
2292
>
VersionedTransitionOffset: *taskOffset,
2293
>
PhysicalTaskStatus: physicalTaskStatusNone,
2294
>
}
2295
>
2296
>
if registrableTask.isPureTask {
2297
>
if skip := n.applySingletonMode(registrableTask, &componentAttr.PureTasks); skip {
tree.go ×3
2299
}
2301
>
sortPureTasks = true
2303
>
if skip := n.applySingletonMode(registrableTask, &componentAttr.SideEffectTasks); skip {
2305
}
2306
>
componentAttr.SideEffectTasks = append(componentAttr.SideEffectTasks, componentTask)
tree.go ×2
2307
}
2308
2310
}
2311
2314
>
slices.SortFunc(componentAttr.PureTasks, comparePureTasks)
2315
>
}
2316
2318
}
2319
2320
func (n *Node) closeTransactionGeneratePhysicalSideEffectTask(
2321
sideEffectTask *persistencespb.ChasmComponentAttributes_Task,
2322
nodePath []string,
2323
archetypeID ArchetypeID,
2325
>
n.backend.AddTasks(&tasks.ChasmTask{
2326
>
WorkflowKey: n.backend.GetWorkflowKey(),
2327
>
VisibilityTimestamp: sideEffectTask.ScheduledTime.AsTime(),
2328
>
Destination: sideEffectTask.Destination,
2329
>
Category: taskCategory(sideEffectTask),
2330
>
Info: &persistencespb.ChasmTaskInfo{
2331
>
ComponentInitialVersionedTransition: n.serializedNode.Metadata.InitialVersionedTransition,
2332
>
ComponentLastUpdateVersionedTransition: n.serializedNode.Metadata.LastUpdateVersionedTransition,
2333
>
Path: nodePath,
2334
>
TypeId: sideEffectTask.TypeId,
2335
>
Data: sideEffectTask.Data,
2336
>
ArchetypeId: archetypeID,
2337
>
TaskVersionedTransition: sideEffectTask.VersionedTransition,
2338
>
TaskVersionedTransitionOffset: sideEffectTask.VersionedTransitionOffset,
2339
>
},
2340
>
})
2341
>
sideEffectTask.PhysicalTaskStatus = physicalTaskStatusCreated
2342
>
}
2343
2344
func (n *Node) closeTransactionGeneratePhysicalPureTask(
2345
firstPureTask *persistencespb.ChasmComponentAttributes_Task,
2346
firstTaskNode *Node,
2347
archetypeID ArchetypeID,
2349
>
if firstPureTask == nil {
2351
>
return nil
2352
>
}
2353
2355
>
n.backend.DeleteCHASMPureTasks(firstPureTaskScheduledTime)
2356
>
2357
>
if firstPureTask.PhysicalTaskStatus == physicalTaskStatusCreated {
2359
>
}
2360
2362
>
WorkflowKey: n.backend.GetWorkflowKey(),
2363
>
VisibilityTimestamp: firstPureTaskScheduledTime,
2364
>
ArchetypeID: archetypeID,
2365
>
})
2366
>
2367
>
// We need to persist the task status change as well, so add the node
2368
>
// to the list of updated nodes.
2369
>
// However, since task status is a cluster local field, we don't really
2370
>
// update LastUpdateVersionedTransition for this node, and the change won't be replicated.
2371
>
firstPureTask.PhysicalTaskStatus = physicalTaskStatusCreated
2372
>
encodedPath, err := firstTaskNode.getEncodedPath()
2373
>
if err != nil {
2374
return err
2375
}
2377
>
return nil
2378
}
2379
2380
// resolveDeferredPointers resolves all deferred pointers in the tree.
2381
// Returns error if any deferred pointer cannot be resolved, as deferred pointers
2382
// cannot be persisted after transaction close.
2384
>
for _, node := range n.andAllChildren() {
2385
>
if node.value == nil || !node.isComponent() {
2387
}
2388
2390
>
if field.err != nil {
2391
return field.err
2392
}
2393
2395
>
continue
2396
}
2397
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 {
2411
>
if !targetNode.isAncestorOf(node) {
2413
>
"pointer target is not an ancestor of component at path %v",
2414
>
node.path(),
2415
>
)
2416
>
}
2417
}
2418
case proto.Message:
2419
resolvedPath, err = n.dataNodePath(value)
2420
default:
2421
err = softassert.UnexpectedInternalErr(
2422
n.logger,
2423
"unable to create a deferred pointer for values of type",
2424
fmt.Errorf("%T", value))
2425
}
2428
>
n.logger,
2429
>
"failed to resolve deferred pointer during transaction close",
2430
>
err)
2431
>
}
2432
2433
// Update the field to be a regular pointer, reusing the existing serializedNode,
2434
// and update the serializedNode's value.
2436
>
newInternal.node = internal.node
2437
>
newInternal.node.setValue(resolvedPath)
2438
>
internalV.Set(reflect.ValueOf(newInternal))
2439
}
2440
}
2441
}
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.
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
}
2456
>
}
2459
>
copy(childPath, path)
2460
>
childPath[len(path)] = child.nodeName
2461
>
if !walk(childPath, child) {
2462
return false
2463
}
2464
}
2466
}
2468
}
2469
}
2470
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 {
2480
>
}
2482
n.systemMutation.DeletedNodes = make(map[string]struct{})
2483
}
2484
2486
>
if len(n.immediatePureTasks) != 0 {
2487
// n.immediatePureTasks should already be empty after executeImmediatePureTasks()
2488
// unless there's an error.
2489
n.immediatePureTasks = make(map[any][]taskWithAttributes)
2490
}
2491
2493
n.pendingRequestLinks = make(map[any]map[string][]*commonpb.Link)
2494
}
2496
n.pendingUserMetadata = make(map[any]*sdkpb.UserMetadata)
2497
}
2498
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
2507
// Snapshot returns all nodes in the tree that have been modified after the given min versioned transition.
2508
// A nil exclusiveMinVT will be treated as the same as the zero versioned transition and returns all nodes in the tree.
2509
// This method should only be invoked on root CHASM node when IsDirty() is false.
2510
func (n *Node) Snapshot(
2511
exclusiveMinVT *persistencespb.VersionedTransition,
2513
>
if !softassert.That(n.logger, n.parent == nil, "chasm.Snapshot() should only be called on the root node") {
2514
panic(fmt.Sprintf("chasm.Snapshot() called on child node: %+v", n))
2515
}
2516
2517
// TODO: add assertion on IsDirty() once implemented
2518
2520
>
n.snapshotInternal(exclusiveMinVT, nodes)
2521
>
2522
>
return NodesSnapshot{
2523
>
Nodes: nodes,
2524
>
}
2525
}
2526
2527
func (n *Node) snapshotInternal(
2528
exclusiveMinVT *persistencespb.VersionedTransition,
2529
nodes map[string]*persistencespb.ChasmNode,
2531
>
if n == nil {
2532
return
2533
}
2534
2535
>
if transitionhistory.Compare(n.serializedNode.Metadata.LastUpdateVersionedTransition, exclusiveMinVT) > 0 {
tree.go ×5
2537
>
if !softassert.That(n.logger, err == nil, "chasm path encoding should always succeed on clean tree") {
2538
panic(fmt.Sprintf("failed to encode chasm path on clean tree: %v", err))
2539
}
2541
}
2542
2545
>
exclusiveMinVT,
2546
>
nodes,
2547
>
)
2548
>
}
2549
}
2550
2551
// PartitionedSnapshot returns the tree's state split into two parts:
2552
// - A NodesSnapshot with cluster-local fields (physical task statuses) zeroed, safe to
2553
// upload to object storage or replicate to another cluster.
2554
// - A ChasmLocalState capturing the extracted cluster-local fields, keyed by encoded
2555
// node path. Only nodes that carry such metadata are present.
2556
//
2557
// The returned snapshot has the same node keys as Snapshot would: PartitionedSnapshot only
2558
// zeroes field values, it never adds or removes nodes. The live in-memory tree is left
2559
// untouched: nodes whose cluster-local fields are zeroed are deep-copied first, since
2560
// Snapshot returns the tree's live node references.
2561
//
2562
// The returned ChasmLocalState has an empty Nodes map when no node carries cluster-local
2563
// fields; MergeClusterLocalState treats an empty (or nil) state as a no-op.
2564
func (n *Node) PartitionedSnapshot(
2565
exclusiveMinVT *persistencespb.VersionedTransition,
2567
>
snapshot := n.Snapshot(exclusiveMinVT)
2568
>
localState := &persistencespb.ChasmLocalState{
2569
>
Nodes: make(map[string]*persistencespb.ChasmNodeLocalState),
2570
>
}
2571
>
for path, node := range snapshot.Nodes {
2572
>
componentAttr := node.GetMetadata().GetComponentAttributes()
2573
>
if componentAttr == nil {
2574
continue
2575
}
2577
>
continue
2578
}
2579
// Deep-copy only the metadata (where physical task statuses live); the Data payload is
2580
// read-only in a snapshot, so share its pointer rather than copying component payloads.
2581
>
clean := &persistencespb.ChasmNode{Metadata: proto.CloneOf(node.GetMetadata()), Data: node.GetData()}
tree.go ×3
2582
>
cleanAttr := clean.GetMetadata().GetComponentAttributes()
2583
>
localState.Nodes[path] = &persistencespb.ChasmNodeLocalState{
2584
>
SideEffectTaskStatuses: extractAndZeroTaskStatuses(cleanAttr.SideEffectTasks),
2585
>
PureTaskStatuses: extractAndZeroTaskStatuses(cleanAttr.PureTasks),
2586
>
}
2587
>
snapshot.Nodes[path] = clean
2588
}
2590
}
2591
2592
// extractAndZeroTaskStatuses records each task's physical task status in order and zeroes
2593
// it in place. The caller must pass tasks from a node copy, not the live tree.
2594
>
func extractAndZeroTaskStatuses(taskList []*persistencespb.ChasmComponentAttributes_Task) []int32 {
tree.go ×3
2595
>
if len(taskList) == 0 {
2596
return nil
2597
}
2599
>
for i, t := range taskList {
2600
>
statuses[i] = t.PhysicalTaskStatus
2601
>
t.PhysicalTaskStatus = physicalTaskStatusNone
2602
>
}
2603
>
return statuses
2604
}
2605
2606
// ClusterLocalStateMergeResult reports, per direction, how many nodes had a task/status count
2607
// mismatch during MergeClusterLocalState. The two directions differ in significance, so they are
2608
// tracked separately rather than as a single count.
2609
type ClusterLocalStateMergeResult struct {
2610
// NodesWithUncoveredTasks counts nodes that had more tasks than stored statuses. The extra
2611
// tasks keep their zeroed status (physicalTaskStatusNone) and get a physical task created on
2612
// the next transaction. Benign and self-healing — typically the writer's captured state was
2613
// slightly behind the authoritative snapshot.
2614
NodesWithUncoveredTasks int
2615
// NodesWithExtraStatuses counts nodes that had more stored statuses than tasks. The surplus
2616
// statuses have no task to apply to and are dropped. Suspicious: the stored state referenced
2617
// tasks absent from the authoritative snapshot, which can indicate cluster divergence (e.g.
2618
// split-brain). Detecting and resolving true divergence belongs to the replication conflict
2619
// path; this count is a diagnostic signal, not the resolution.
2620
NodesWithExtraStatuses int
2621
}
2622
2623
// MergeClusterLocalState restores cluster-local metadata into the snapshot, inverting the
2624
// extraction performed by PartitionedSnapshot. Nodes present in both the snapshot and the
2625
// state are updated; nodes in the state but not the snapshot are silently skipped (the node
2626
// may have been deleted). Statuses are matched to tasks by position; a length mismatch applies
2627
// only the overlapping prefix. It returns per-direction counts of nodes whose status count didn't
2628
// match the task count (see ClusterLocalStateMergeResult), so callers can react to a (usually
2629
// stale-data) merge and escalate the suspicious direction.
2630
//
2631
// The merge performs no version/ordering checks; callers must apply it only against a final local
2632
// tree (e.g. defer until the execution is completed and its close version has caught up to the
2633
// source), so a length mismatch signals real divergence rather than normal replication lag.
2634
>
func (s *NodesSnapshot) MergeClusterLocalState(state *persistencespb.ChasmLocalState) ClusterLocalStateMergeResult {
tree.go ×2
2635
>
var result ClusterLocalStateMergeResult
2636
>
for path, nodeState := range state.GetNodes() {
2638
>
if !ok {
2640
}
2642
>
if componentAttr == nil {
2643
continue
2644
}
2645
>
seDiff := mergeTaskStatuses(componentAttr.SideEffectTasks, nodeState.GetSideEffectTaskStatuses())
tree.go ×4
2646
>
pureDiff := mergeTaskStatuses(componentAttr.PureTasks, nodeState.GetPureTaskStatuses())
2647
>
if seDiff > 0 || pureDiff > 0 {
2649
>
}
2652
>
}
2653
}
2655
}
2656
2657
// mergeTaskStatuses applies statuses to tasks by position and returns len(taskList) - len(statuses)
2658
// (>0: extra tasks left zeroed; <0: surplus statuses dropped).
2659
>
func mergeTaskStatuses(taskList []*persistencespb.ChasmComponentAttributes_Task, statuses []int32) int {
tree.go ×4
2660
>
for i := 0; i < len(taskList) && i < len(statuses); i++ {
2661
>
taskList[i].PhysicalTaskStatus = statuses[i]
2662
>
}
2663
>
return len(taskList) - len(statuses)
2664
}
2665
2666
// ApplySystemMutation should only used by internal persistence layer logic to force apply
2667
// cluster specific chasm tree changes.
2668
// DO NOT USE if you don't know why this method is introduced.
2669
func (n *Node) ApplySystemMutation(
2670
mutation NodesMutation,
2671
) error {
2672
if err := n.applyDeletions(mutation.DeletedNodes, true); err != nil {
2673
return err
2674
}
2675
2676
return n.applyUpdates(mutation.UpdatedNodes, true)
2677
}
2678
2679
// ApplyMutation is used by replication stack to apply node
2680
// mutations from the source cluster.
2681
//
2682
// NOTE: It will be an error if UpdatedNodes and DeletedNodes have overlapping keys,
2683
// as the CHASM tree does not have enough information to tell if the deletion happens
2684
// before or after the update.
2685
func (n *Node) ApplyMutation(
2686
mutation NodesMutation,
2688
>
if err := n.applyDeletions(mutation.DeletedNodes, false); err != nil {
2689
return err
2690
}
2691
2693
return err
2694
}
2695
2696
// For replication case, we only update the search attributes and memo
2697
// but not force updating the visibility component itself to generate a task.
2698
//
2699
// This is because the visibility component is already force updated in the active
2700
// cluster and that forced update will be replicated as well. Standby cluster
2701
// only needs to track the current SA and memo to prevent generating an unnecessary
2702
// visibility component update & task if there is a failover.
2703
//
2704
// TODO: combine this with the logic in CloseTransactionForceUpdateVisibility
2705
// right that force update logic only applies to the active cluster.
2707
>
rootComponent, err := n.root().Component(immutableContext, ComponentRef{})
2708
>
if err != nil {
2709
return err
2710
}
2712
>
if ok {
2713
>
saSlice := saProvider.SearchAttributes(immutableContext)
2714
>
n.currentSA = searchAttributeKeyValuesToMap(saSlice)
2715
>
}
2716
>
memoProvider, ok := rootComponent.(VisibilityMemoProvider)
2717
>
if ok {
2718
>
n.currentMemo = proto.Clone(memoProvider.Memo(immutableContext))
2719
>
}
2720
2722
}
2723
2724
// ApplySnapshot is used by replication stack to apply node
2725
// snapshot from the source cluster.
2726
//
2727
// If we simply substituting the entire CHASM tree, we will be
2728
// forced to close the transaction as snapshot and potentially
2729
// write extra data to persistence.
2730
// This method will instead figure out the mutations needed to
2731
// bring the current tree to the be the same as the snapshot,
2732
// thus allowing us to close the transaction as mutation.
2733
func (n *Node) ApplySnapshot(
2734
incomingSnapshot NodesSnapshot,
2736
>
currentSnapshot := n.Snapshot(nil)
2737
>
2738
>
mutation := NodesMutation{
2739
>
UpdatedNodes: make(map[string]*persistencespb.ChasmNode),
2740
>
DeletedNodes: make(map[string]struct{}),
2741
>
}
2742
>
2743
>
for encodedPath := range currentSnapshot.Nodes {
2744
>
if _, ok := incomingSnapshot.Nodes[encodedPath]; !ok {
2745
>
mutation.DeletedNodes[encodedPath] = struct{}{}
2746
>
}
2747
}
2748
2751
>
if !ok {
2752
>
mutation.UpdatedNodes[encodedPath] = incomingNode
2753
>
continue
2754
}
2755
2757
>
currentNode.Metadata.LastUpdateVersionedTransition,
2758
>
incomingNode.Metadata.LastUpdateVersionedTransition,
2759
>
) != 0 {
2760
>
mutation.UpdatedNodes[encodedPath] = incomingNode
2761
>
}
2762
}
2763
2765
}
2766
2767
func (n *Node) applyDeletions(
2768
deletedNodes map[string]struct{},
2769
isSystemUpdates bool,
2771
>
for encodedPath := range deletedNodes {
2773
>
if err != nil {
2774
return err
2775
}
2776
2778
>
if !ok {
2780
>
// This could happen when:
2781
>
// - If the mutations passed in include changes
2782
>
// older than the current state of the tree.
2783
>
// - We are already applied the deletion on a parent node.
2784
>
continue
2785
}
2786
2789
>
// This can happen when:
2790
>
// 1. CHASM framework is disabled in source cluster and sends an
2791
>
// empty snapshot to the standby cluster. If the standby cluster
2792
>
// has a non-empty chasm tree, the root node will be marked for
2793
>
// deletion and we will lose archetype information for the execution,
2794
>
// and hit other undefined issues when root is deleted.
2795
>
// Disabled CHASM framework itself is already an undefined situation
2796
>
// for non-workflow chasm executions, and we are ok with not deleting
2797
>
// the root node.
2798
>
//
2799
>
// 2. CHASM is enabled but the execution is a workflow which doesn't
2800
>
// have any chasm nodes. In this case, again an empty snapshot will be sent to
2801
>
// standby cluster.
2802
>
// In this case, we can actually choose to delete the root itself because empty
2803
>
// chasm tree is assume to be a Workflow. However, given chasm workflow component's
2804
>
// state is an empty proto, skipping deletion is fine as well. All other child nodes
2805
>
// will still be deleted.
2806
>
continue
2807
}
2808
2810
>
if err := node.delete(isSystemUpdates); err != nil {
2811
return err
2812
}
2814
}
2815
2817
}
2818
2819
func (n *Node) applyUpdates(
2820
updatedNodes map[string]*persistencespb.ChasmNode,
2821
isSystemUpdates bool,
2823
>
for encodedPath, updatedNode := range updatedNodes {
2825
>
if err != nil {
2826
return err
2827
}
2828
2830
>
if !ok {
2832
>
newNode := n.setSerializedNode(path, encodedPath, updatedNode)
2833
>
newNode.resetTaskStatus()
2834
>
n.clearAncestorNodeValues(newNode.parent)
2835
>
if isSystemUpdates {
2836
n.systemMutation.UpdatedNodes[encodedPath] = newNode.serializedNode
2837
delete(n.systemMutation.DeletedNodes, encodedPath)
2839
>
n.mutation.UpdatedNodes[encodedPath] = newNode.serializedNode
2840
>
delete(n.mutation.DeletedNodes, encodedPath)
2841
>
}
2842
>
continue
2843
}
2844
2845
// An empty node may be created when child update is applied before the parent,
2846
// in which case node.serializedNode will be nil.
2848
>
node.serializedNode.Metadata.LastUpdateVersionedTransition,
2849
>
updatedNode.Metadata.LastUpdateVersionedTransition,
2850
>
) != 0 {
2851
>
localComponentAttr := node.serializedNode.GetMetadata().GetComponentAttributes()
2852
>
updatedComponentAttr := updatedNode.GetMetadata().GetComponentAttributes()
2853
>
if localComponentAttr != nil && updatedComponentAttr != nil {
2854
>
n.carryOverTaskStatus(
2855
>
localComponentAttr.SideEffectTasks,
2856
>
updatedComponentAttr.SideEffectTasks,
2857
>
compareSideEffectTasks,
2858
>
)
2859
>
n.carryOverTaskStatus(
2860
>
localComponentAttr.PureTasks,
2861
>
updatedComponentAttr.PureTasks,
2862
>
comparePureTasks,
2863
>
)
2864
>
}
2865
2867
n.systemMutation.UpdatedNodes[encodedPath] = updatedNode
2868
delete(n.systemMutation.DeletedNodes, encodedPath)
2870
>
n.mutation.UpdatedNodes[encodedPath] = updatedNode
2871
>
delete(n.mutation.DeletedNodes, encodedPath)
2872
>
}
2873
>
node.setValue(nil)
2874
>
node.setValueState(valueStateNeedDeserialize)
2875
>
node.serializedNode = updatedNode
2876
>
n.clearAncestorNodeValues(node.parent)
2877
}
2878
}
2879
2881
}
2882
2884
>
for _, node := range n.andAllChildren() {
2885
>
// Only reset task status here, the actual task generation will be done when
2886
>
// CloseTransaction() is called to persist the changes.
2887
>
if reset := node.resetTaskStatus(); !reset {
2888
>
continue
2889
}
2890
2892
>
if err != nil {
2893
return err
2894
}
2895
2896
// Task status is a cluster local field and changes to it doesn't need to be replicated.
2897
// Recording changes in systemMutation so that:
2898
// 1. it can be persisted.
2899
// 2. n.IsStateDirty() can still return false so that mutable state's transition history
2900
// won't be updated.
2902
}
2903
2905
}
2906
2908
>
if n.serializedNode == nil || n.serializedNode.GetMetadata() == nil {
2909
return false
2910
}
2911
2913
>
if componentAttr == nil {
2915
>
}
2916
2918
>
for _, componentTasks := range [][]*persistencespb.ChasmComponentAttributes_Task{
2919
>
componentAttr.PureTasks,
2920
>
componentAttr.SideEffectTasks,
2921
>
} {
2922
>
for _, t := range componentTasks {
2924
>
reset = true
2925
>
}
2926
>
t.PhysicalTaskStatus = physicalTaskStatusNone
2927
}
2928
}
2929
2931
}
2932
2934
>
if n.encodedPath != nil {
2936
>
}
2938
>
if err == nil {
2939
>
n.encodedPath = &encodePath
2940
>
}
2941
>
return encodePath, err
2942
}
2943
2945
>
if n.parent == nil {
2946
>
return []string{}
2947
>
}
2948
2950
}
2951
2952
func (n *Node) findNode(
2953
path []string,
2955
>
if len(path) == 0 {
2956
>
return n, true
2957
>
}
2958
2960
>
childNode, ok := n.children[childName]
2961
>
if !ok {
2963
>
}
2965
}
2966
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.
2970
>
current := descendant.parent
2971
>
for current != nil {
2973
>
return true
2974
>
}
2976
}
2978
}
2979
2981
>
for _, childNode := range n.children {
2983
return err
2984
}
2985
}
2986
2987
// If a parent is about to be removed, it must not have any children.
2988
>
softassert.That(n.logger, len(n.children) == 0, "children must be empty when node is removed")
tree.go ×6
2989
>
2990
>
if n.parent != nil {
2991
>
delete(n.parent.children, n.nodeName)
2992
>
}
2993
2994
// Set value to nil which also deletes the value from valueToNode map.
2996
>
2997
>
encodedPath, err := n.getEncodedPath()
2998
>
if err != nil {
2999
return err
3000
}
3001
3002
// Only record the deletion if the node was previously persisted.
3003
//
3004
// TODO: consider remove entries from UpdatedNodes map as well
3005
// if the same node is updated and then deleted in the same transaction.
3006
//
3007
// That's not a problem today though and DeletedNodes entries are always added
3008
// before UpdatedNodes entires.
3009
// - For active logic, DeletedNodes are added upon syncSubComponents(),
3010
// and UpdatedNodes are added when closing transaction and serializing nodes.
3011
// - For standby replication logic, mutable state calls ApplyMutation() twice,
3012
// first with a deletion only mutation for tombstone nodes, and then an
3013
// update only mutation.
3016
n.systemMutation.DeletedNodes[encodedPath] = struct{}{}
3018
>
n.mutation.DeletedNodes[encodedPath] = struct{}{}
3019
>
}
3020
}
3021
3023
>
3024
>
return nil
3025
}
3026
3028
>
if !n.isComponent() {
3030
>
}
3031
3033
>
for _, task := range componentAttr.GetPureTasks() {
3035
>
}
3037
delete(n.taskValueCache, task.Data)
3038
}
3039
}
3040
3041
// IsDirty returns true if any node in the tree has been modified,
3042
// and need to be persisted in DB.
3043
// The result will be reset to false after a call to CloseTransaction().
3045
>
if n.IsStateDirty() {
3047
>
}
3048
3049
>
return len(n.systemMutation.UpdatedNodes) > 0 || len(n.systemMutation.DeletedNodes) > 0
tree.go ×2
3050
}
3051
3052
// IsStateDirty returns true if any node in the tree has USER DATA modified,
3053
// which need to be persisted to DB AND replicated to other clusters.
3054
// The result will be reset to false after a call to CloseTransaction().
3056
>
return n.subtreeIsDirty ||
3057
>
len(n.mutation.UpdatedNodes) > 0 ||
3058
>
len(n.mutation.DeletedNodes) > 0
3059
>
}
3060
3061
func (n *Node) IsStale(
3062
ref ComponentRef,
3064
>
// The point of this method to access the private executionLastUpdateVT field in componentRef,
3065
>
// and avoid exposing it in the public CHASM interface.
3066
>
if ref.executionLastUpdateVT == nil {
3068
>
}
3069
3071
>
n.backend.GetExecutionInfo().TransitionHistory,
3072
>
ref.executionLastUpdateVT,
3073
>
)
3074
}
3075
3076
func (n *Node) Terminate(
3077
request TerminateComponentRequest,
3079
>
if n.parent != nil {
3080
return softassert.UnexpectedInternalErr(
3081
n.logger,
3082
"Terminate should only be called on the root node",
3083
fmt.Errorf("node path: %v", n.path()),
3084
)
3085
}
3086
3088
>
component, err := n.Component(mutableContext, ComponentRef{})
3089
>
if err != nil {
3090
return err
3091
}
3093
>
if !ok {
3094
return softassert.UnexpectedInternalErr(
3095
n.logger,
3096
"root node must implement RootComponent interface",
3097
fmt.Errorf("component type: %T", component),
3098
)
3099
}
3100
3102
>
if err != nil {
3103
return err
3104
}
3105
3107
>
return nil
3108
}
3109
3110
// SetDeleteAfterClose suppresses the close visibility task when an execution is being
3111
// terminated as part of a delete operation. Must be called before a [Terminate] call, like in DeleteExecution.
3113
>
n.deleteAfterClose = deleteAfterClose
3114
>
}
3115
3116
// ArchetypeID returns the framework's internal ID for the root component's fully qualified name.
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.
3123
// Deprecated: use ArchetypeID() instead, this method will be removed.
3124
func (n *Node) Archetype() (Archetype, error) {
3125
archetypeID := n.ArchetypeID()
3126
3127
fqn, ok := n.registry.ComponentFqnByID(archetypeID)
3128
if !ok {
3129
return "", softassert.UnexpectedInternalErr(
3130
n.logger,
3131
"unknown archetype id",
3132
fmt.Errorf("%d", archetypeID))
3133
}
3134
3135
return Archetype(fqn), nil
3136
}
3137
3139
>
if n.parent == nil {
3140
>
return n
3141
>
}
3143
}
3144
3145
// isComponentTaskExpired returns true when the task's scheduled time is equal
3146
// or before the reference time. The caller should also make sure to account
3147
// for skew between the physical task queue and the database by adjusting
3148
// referenceTime in advance.
3149
func isComponentTaskExpired(
3150
referenceTime time.Time,
3151
task *persistencespb.ChasmComponentAttributes_Task,
3153
>
if task.ScheduledTime == nil {
3154
return false
3155
}
3156
3157
>
scheduledTime := task.ScheduledTime.AsTime().Truncate(common.ScheduledTaskMinPrecision)
tree.go ×15
3158
>
referenceTime = referenceTime.Truncate(common.ScheduledTaskMinPrecision)
3159
>
3160
>
return !scheduledTime.After(referenceTime)
3161
}
3162
3163
// EachPureTask runs the callback for all expired/runnable pure tasks within the
3164
// CHASM tree (including invalid tasks). The CHASM tree is left untouched, even
3165
// if invalid tasks are detected (these are cleaned up as part of transaction
3166
// close).
3167
func (n *Node) EachPureTask(
3168
referenceTime time.Time,
3169
callback func(handler NodePureTask, taskAttributes TaskAttributes, taskInstance any) (bool, error),
3171
>
chasmContext := NewContext(context.Background(), n)
3172
>
3173
>
// Because tree structure may change during the processing,
3174
>
// we first gather all nodes that have pure tasks that are ready for execution.
3175
>
var componentToProcess []any
3176
>
for _, node := range n.andAllChildren() {
3177
>
// Skip nodes that aren't serialized yet.
3178
>
if node.serializedNode == nil || node.serializedNode.Metadata == nil {
3179
continue
3180
}
3181
3183
>
// Skip nodes that aren't components.
3184
>
if componentAttr == nil {
3185
continue
3186
}
3187
3189
continue
3190
}
3191
3193
>
continue
3194
}
3195
3196
// This component node as a pure task that's ready to execute
3198
>
if err != nil {
3199
return err
3200
}
3201
3203
}
3204
3206
>
3207
>
// Node get deleted when previous pure tasks of other components are executed.
3208
>
node, ok := n.valueToNode[component]
3209
>
if !ok {
3210
>
continue
3211
}
3212
3214
>
3215
>
for _, task := range componentAttr.GetPureTasks() {
3216
>
if !isComponentTaskExpired(referenceTime, task) {
3217
break
3218
}
3219
3220
// Node get deleted when previous pure tasks of the same component are executed.
3221
// e.g. via a (parent) pointer.
3223
>
if !ok {
3224
>
break
3225
}
3226
3228
>
if err != nil {
3229
return err
3230
}
3231
3233
>
ScheduledTime: task.ScheduledTime.AsTime(),
3234
>
Destination: task.Destination,
3235
>
}
3236
>
3237
>
executed, err := callback(node, taskAttributes, taskInstance)
3238
>
if err != nil {
3239
return err
3240
}
3241
3243
>
if err := n.syncSubComponents(); err != nil {
3244
return err
3245
}
3246
}
3247
3248
// Processed task should become invalid and will be removed upon CloseTransaction().
3249
3250
// TODO: Add a validation for that and return an internal error if tasks is still valid after processing.
3251
// Alternatively, remove task from PureTasks slice after processing, but that requires persisting the
3252
// task changes as well even if the component itself is not changed.
3253
}
3254
}
3255
3257
}
3258
3259
func newNode(
3260
base *nodeBase,
3261
parent *Node,
3262
nodeName string,
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 {
tree.go ×2
3273
>
if cmpResult := transitionhistory.Compare(a.VersionedTransition, b.VersionedTransition); cmpResult != 0 {
3275
>
}
3277
}
3278
3280
>
if cmpResult := a.ScheduledTime.AsTime().Compare(b.ScheduledTime.AsTime()); cmpResult != 0 {
3282
>
}
3283
3285
}
3286
3287
func (n *Node) carryOverTaskStatus(
3288
sourceTasks, targetTasks []*persistencespb.ChasmComponentAttributes_Task,
3289
compareFn func(a, b *persistencespb.ChasmComponentAttributes_Task) int,
3291
>
sourceIdx, targetIdx := 0, 0
3292
>
for sourceIdx < len(sourceTasks) && targetIdx < len(targetTasks) {
3294
>
targetTask := targetTasks[targetIdx]
3295
>
3296
>
switch compareFn(sourceTask, targetTask) {
3298
>
// Task match, carry over status.
3299
>
targetTask.PhysicalTaskStatus = sourceTask.PhysicalTaskStatus
3300
>
// Use existing task data to avoid taskValueCache miss, since the cache uses
3301
>
// *DataBlob as the key.
3302
>
// Otherwise we have to clear cache for all tasks in the node, and re-deserialize
3303
>
// tasks later.
3304
>
targetTask.Data = sourceTask.Data
3305
>
sourceIdx++
3306
>
targetIdx++
3308
>
// Source task has a smaller key, meaning the task has been deleted.
3309
>
// Move on to the next source task.
3310
>
sourceIdx++
3311
>
delete(n.taskValueCache, sourceTask.Data)
3313
>
// Source task has a larger key, meaning there's a new task inserted.
3314
>
// Sanitize incoming task status.
3315
>
targetTask.PhysicalTaskStatus = physicalTaskStatusNone
3316
>
targetIdx++
3317
}
3318
}
3319
3320
// Sanitize incoming task status for remaining tasks.
3323
>
}
3326
>
}
3327
}
3328
3329
func taskCategory(
3330
task *persistencespb.ChasmComponentAttributes_Task,
3332
>
if task.TypeId == visibilityTaskTypeID {
3334
>
}
3335
3338
>
}
3339
3341
>
task.ScheduledTime.AsTime().Equal(TaskScheduledTimeImmediate) {
3342
>
return tasks.CategoryTransfer
3343
>
}
3345
}
3346
3347
func (n *Node) deserializeTaskWithCache(
3348
registrableTask *RegistrableTask,
3349
taskBlob *commonpb.DataBlob,
3351
>
if cachedValue, ok := n.taskValueCache[taskBlob]; ok {
3353
>
}
3354
3356
>
if err != nil {
3357
return reflect.Value{}, err
3358
}
3359
3361
>
return taskValue, nil
3362
}
3363
3364
func (n *Node) serializeTaskWithCache(
3365
registrableTask *RegistrableTask,
3366
taskValue reflect.Value,
3368
>
taskBlob, err := serializeTask(registrableTask, taskValue)
3369
>
if err != nil {
3370
return nil, err
3371
}
3372
3374
>
return taskBlob, nil
3375
}
3376
3377
func deserializeTask(
3378
registrableTask *RegistrableTask,
3379
taskBlob *commonpb.DataBlob,
3381
>
if registrableTask.goType.AssignableTo(protoMessageT) {
3383
>
if err != nil {
3384
return reflect.Value{}, err
3385
}
3387
}
3388
3390
>
if taskGoType.Kind() == reflect.Pointer {
3392
>
}
3394
>
3395
>
// At this point taskGoType is guaranteed to be a struct and
3396
>
// taskValue is a pointer to struct.
3397
>
3398
>
defer func() {
3399
>
if retErr == nil && registrableTask.goType.Kind() == reflect.Struct {
3401
>
}
3402
}()
3403
3405
>
return taskValue, nil
3406
>
}
3407
3408
// TODO: consider pre-calculating the proto field num when registring the task type.
3409
3410
protoMessageFound := false
3411
for i := 0; i < taskGoType.NumField(); i++ {
3412
fieldV := taskValue.Elem().Field(i)
3413
fieldT := taskGoType.Field(i).Type
3414
if !fieldT.AssignableTo(protoMessageT) {
3415
continue
3416
}
3417
3418
if protoMessageFound {
3419
return reflect.Value{}, serviceerror.NewInternal("only one proto field allowed in task struct")
3420
}
3421
protoMessageFound = true
3422
3423
value, err := unmarshalProto(taskBlob, fieldT)
3424
if err != nil {
3425
return reflect.Value{}, err
3426
}
3427
3428
fieldV.Set(value)
3429
}
3430
3431
return taskValue, nil
3432
}
3433
3434
func serializeTask(
3435
registrableTask *RegistrableTask,
3436
taskValue reflect.Value,
3438
>
protoValue, ok := taskValue.Interface().(proto.Message)
3439
>
if ok {
3441
>
}
3442
3444
>
3445
>
// Handle pointer to struct.
3446
>
if taskGoType.Kind() == reflect.Pointer {
3447
taskGoType = taskGoType.Elem()
3448
taskValue = taskValue.Elem()
3449
}
3450
3451
// Handle empty task struct.
3453
>
return encodeChasmBlob(nil)
3454
>
}
3455
3456
// TODO: consider pre-calculating the proto field num when registring the task type.
3457
3458
var blob *commonpb.DataBlob
3459
protoMessageFound := false
3460
for i := 0; i < taskGoType.NumField(); i++ {
3461
fieldV := taskValue.Field(i)
3462
if !fieldV.Type().AssignableTo(protoMessageT) {
3463
continue
3464
}
3465
3466
if protoMessageFound {
3467
return nil, serviceerror.NewInternalf("only one proto field allowed in task struct of type: %v", taskGoType.String())
3468
}
3469
protoMessageFound = true
3470
3471
var err error
3472
blob, err = encodeChasmBlob(fieldV.Interface().(proto.Message))
3473
if err != nil {
3474
return nil, err
3475
}
3476
}
3477
3478
if !protoMessageFound {
3479
return nil, serviceerror.NewInternal("no proto field found in task struct")
3480
}
3481
3482
return blob, nil
3483
}
3484
3485
// ExecutePureTask validates and then executes the given taskInstance against the
3486
// node's component. Executing an invalid task is a no-op (no error returned).
3487
func (n *Node) ExecutePureTask(
3488
baseCtx context.Context,
3489
taskAttributes TaskAttributes,
3490
taskInstance any,
3492
>
defer func() {
3493
>
if retErr == nil {
3494
>
// Mark this node dirty so CloseTransaction cleans up invalid tasks,
3495
>
// including the current one, even if the handler made no state mutations.
3496
>
n.markSubtreeDirty()
3497
>
}
3498
}()
3499
3501
>
if !ok {
3502
return false, fmt.Errorf("unknown task type for task instance goType '%s'", reflect.TypeOf(taskInstance).Name())
3503
}
3504
3506
return false, fmt.Errorf("ExecutePureTask called on a SideEffect task '%s'", registrableTask.fqType())
3507
}
3508
3509
>
progressIntentCtx := newContextWithOperationIntent(baseCtx, OperationIntentProgress)
tree.go ×6
3510
>
validationContext := NewContext(progressIntentCtx, n)
3511
>
3512
>
// Ensure this node's component value is hydrated before execution.
3513
>
if err := n.prepareComponentValue(validationContext); err != nil {
3514
return false, err
3515
}
3516
3517
// Run the task's registered value before execution.
3518
>
valid, err := n.validateTask(validationContext, TaskInvocation{TaskAttributes: taskAttributes}, taskInstance)
tree.go ×6
3519
>
if err != nil {
3521
>
}
3524
>
}
3525
3527
>
component, err := n.Component(executionContext, ComponentRef{})
3528
>
if err != nil {
3529
return false, err
3530
}
3531
3533
>
3534
>
archetypeTag := metrics.ArchetypeTag("")
3535
>
if name, ok := n.registry.ArchetypeDisplayName(n.ArchetypeID()); ok {
3536
>
archetypeTag = metrics.ArchetypeTag(name)
3537
>
}
3538
>
chasmTaskTypeTag := metrics.ChasmTaskTypeTag(registrableTask.fqType())
3539
>
metricsHandler := n.metricsHandler.WithTags(archetypeTag)
3540
>
3541
>
execErr := registrableTask.pureTaskExecuteFn(
3542
>
executionContext,
3543
>
component,
3544
>
taskAttributes,
3545
>
taskInstance,
3546
>
n.registry,
3547
>
)
3548
>
3549
>
metrics.ChasmPureTaskRequests.With(metricsHandler).Record(1, chasmTaskTypeTag)
3550
>
3551
>
if execErr != nil {
3553
>
return true, execErr
3554
>
}
3555
3556
// TODO - a task validator must succeed validation after a task executes
3557
// successfully (without error), otherwise it will generate an infinite loop.
3558
// Check for this case by marking the in-memory task as having executed, which the
3559
// CloseTransaction method will check against.
3560
//
3561
// See: https://github.com/temporalio/temporal/pull/7701#discussion_r2072026993
3562
3564
}
3565
3566
// ValidateSideEffectTask checks whether a side effect task should still be
3567
// executed. Intended for use by standby handlers.
3568
//
3569
// It returns two booleans:
3570
// - isTaskInTree: true if the task's logical counterpart still exists in the
3571
// replicated tree state (node found, InitialVersionedTransition matches, and
3572
// logical task present in SideEffectTasks). A false value here means the
3573
// active cluster has definitively invalidated the task via replication — the
3574
// physical task should be dropped.
3575
// - isValidByComponent: true if the component's own Validate method approves
3576
// the task. Only meaningful when isTaskInTree is true. A false value here
3577
// may be a transient false-negative caused by a code deployment changing
3578
// validation logic without a corresponding state change.
3579
//
3580
// If an error is returned both booleans are false.
3581
func (n *Node) ValidateSideEffectTask(
3582
ctx context.Context,
3583
chasmTask *tasks.ChasmTask,
3585
>
3586
>
taskInfo := chasmTask.Info
3587
>
taskTypeID := taskInfo.TypeId
3588
>
registrableTask, ok := n.registry.TaskByID(taskTypeID)
3589
>
if !ok {
3590
return false, false, softassert.UnexpectedInternalErr(
3591
n.logger,
3592
"unknown task type id",
3593
fmt.Errorf("%d", taskTypeID))
3594
}
3595
3597
return false, false, softassert.UnexpectedInternalErr(
3598
n.logger,
3599
"ValidateSideEffectTask called on a Pure task, task type: ",
3600
fmt.Errorf("%s", registrableTask.fqType()))
3601
}
3602
3604
>
if !ok {
3605
return false, false, nil
3606
}
3607
3608
// node.serializedNode should always be available when running a side effect task.
3610
>
taskInfo.ComponentInitialVersionedTransition,
3611
>
node.serializedNode.Metadata.InitialVersionedTransition,
3612
>
) != 0 {
3614
>
}
3615
3616
// Verify the logical task this physical task was generated from still exists,
3617
// and capture it so we can use its Data pointer for the deserialization cache.
3618
//
3619
// A logical task can be dropped mid-flight (e.g. component paused then unpaused)
3620
// without the physical task being cancelled. Checking existence here prevents
3621
// stale physical tasks from executing after their logical counterpart is gone.
3622
//
3623
// TaskVersionedTransition is unset on physical tasks created before this field
3624
// was added; skip the check in that case to preserve backward compatibility.
3626
>
if taskInfo.TaskVersionedTransition != nil {
3627
componentAttr := node.serializedNode.Metadata.GetComponentAttributes()
3628
for _, t := range componentAttr.GetSideEffectTasks() {
3629
if transitionhistory.Compare(t.VersionedTransition, taskInfo.TaskVersionedTransition) == 0 &&
3630
t.VersionedTransitionOffset == taskInfo.TaskVersionedTransitionOffset {
3631
logicalTask = t
3632
break
3633
}
3634
}
3635
if logicalTask == nil {
3636
return false, false, nil
3637
}
3638
}
3639
3640
// All structural checks passed — the task exists in the tree.
3641
3642
// Component must be hydrated before the task's validator is called.
3643
>
validateCtx := NewContext(newContextWithOperationIntent(ctx, OperationIntentProgress), n)
tree.go ×11
3644
>
if err := node.prepareComponentValue(validateCtx); err != nil {
3645
return false, false, err
3646
}
3647
3649
>
if rec := recover(); rec != nil {
3650
chasmTask.DeserializedTask = reflect.Value{}
3651
panic(rec) //nolint:forbidigo
3652
}
3655
>
}
3656
}()
3657
3659
>
var err error
3660
>
if logicalTask != nil {
3661
// Use the logical task's Data pointer so deserialization shares the
3662
// node's taskValueCache with closeTransactionCleanupInvalidTasks.
3663
// The physical task's taskInfo.Data is a different pointer (freshly
3664
// allocated from the physical task row) and would always miss the cache.
3665
chasmTask.DeserializedTask, err = node.deserializeTaskWithCache(registrableTask, logicalTask.Data)
3667
>
// Backward compatibility: physical task predates TaskVersionedTransition.
3668
>
chasmTask.DeserializedTask, err = deserializeTask(registrableTask, taskInfo.Data)
3669
>
}
3670
>
if err != nil {
3671
return false, false, err
3672
}
3673
}
3674
3676
>
validateCtx,
3677
>
TaskInvocation{
3678
>
TaskAttributes: TaskAttributes{
3679
>
ScheduledTime: chasmTask.GetVisibilityTime(),
3680
>
Destination: chasmTask.Destination,
3681
>
},
3682
>
Attempt: chasmTask.Attempt,
3683
>
},
3684
>
chasmTask.DeserializedTask.Interface(),
3685
>
)
3686
>
return true, isValidByComponent, retErr
3687
}
3688
3689
// ExecuteSideEffectTask executes the given ChasmTask on its associated node
3690
// without holding the execution lock.
3691
//
3692
// WARNING: This method *must not* access the node's properties without first
3693
// locking the execution.
3694
//
3695
// ctx should have a CHASM engine already set.
3696
func (n *Node) ExecuteSideEffectTask(
3697
ctx context.Context,
3698
executionKey ExecutionKey,
3699
chasmTask *tasks.ChasmTask,
3700
validate func(NodeBackend, Context, Component) error,
3702
>
rt, err := n.lookupSideEffectTask(ctx, "ExecuteSideEffectTask", chasmTask)
3703
>
if err != nil {
3704
return err
3705
}
3706
>
return n.invokeSideEffectTaskFn(ctx, rt, executionKey, chasmTask, validate, rt.sideEffectTaskExecuteFn)
task_mock.go ×2
3707
}
3708
3709
// ExecuteSideEffectDiscardTask executes the discard handler for the given ChasmTask. This is called on standby
3710
// clusters when a side effect task has been pending past the discard delay, allowing custom discard behavior
3711
// (e.g., spilling activity tasks to matching).
3712
func (n *Node) ExecuteSideEffectDiscardTask(
3713
ctx context.Context,
3714
executionKey ExecutionKey,
3715
chasmTask *tasks.ChasmTask,
3716
validate func(NodeBackend, Context, Component) error,
3718
>
rt, err := n.lookupSideEffectTask(ctx, "ExecuteSideEffectDiscardTask", chasmTask)
3719
>
if err != nil {
3720
return err
3721
}
3722
>
return n.invokeSideEffectTaskFn(ctx, rt, executionKey, chasmTask, validate, rt.sideEffectTaskDiscardFn)
tree.go ×3
3723
}
3724
3725
func (n *Node) lookupSideEffectTask(
3726
ctx context.Context,
3727
callerName string,
3728
chasmTask *tasks.ChasmTask,
3730
>
if engineFromContext(ctx) == nil {
3731
return nil, serviceerror.NewInternal("no CHASM engine set on context")
3732
}
3733
3735
>
registrableTask, ok := n.registry.TaskByID(taskTypeID)
3736
>
if !ok {
3737
return nil, softassert.UnexpectedInternalErr(
3738
n.logger,
3739
"unknown task type id",
3740
fmt.Errorf("%d", taskTypeID))
3741
}
3743
return nil, softassert.UnexpectedInternalErr(
3744
n.logger,
3745
callerName+" called on a Pure task",
3746
fmt.Errorf("%s", registrableTask.fqType()))
3747
}
3749
}
3750
3751
func (n *Node) invokeSideEffectTaskFn(
3752
ctx context.Context,
3753
registrableTask *RegistrableTask,
3754
executionKey ExecutionKey,
3755
chasmTask *tasks.ChasmTask,
3756
validate func(NodeBackend, Context, Component) error,
3757
taskFn func(context.Context, ComponentRef, TaskAttributes, any) error,
3759
>
taskInfo := chasmTask.Info
3760
>
3761
>
defer func() {
3762
>
if rec := recover(); rec != nil {
3763
chasmTask.DeserializedTask = reflect.Value{}
3764
panic(rec) //nolint:forbidigo
3765
}
3768
>
}
3769
}()
3770
3772
>
var err error
3773
>
// TODO: Change physical side effect task to reference logical task and
3774
>
// then use deserializeTaskWithCache as well.
3775
>
chasmTask.DeserializedTask, err = deserializeTask(registrableTask, taskInfo.Data)
3776
>
if err != nil {
3777
return err
3778
}
3779
}
3781
>
3782
>
taskAttributes := TaskAttributes{
3783
>
ScheduledTime: chasmTask.GetVisibilityTime(),
3784
>
Destination: chasmTask.Destination,
3785
>
}
3786
>
3787
>
ref := ComponentRef{
3788
>
ExecutionKey: executionKey,
3789
>
archetypeID: ArchetypeID(taskInfo.GetArchetypeId()),
3790
>
executionLastUpdateVT: taskInfo.ComponentLastUpdateVersionedTransition,
3791
>
componentPath: taskInfo.Path,
3792
>
componentInitialVT: taskInfo.ComponentInitialVersionedTransition,
3793
>
3794
>
// Validate the Ref only once it is accessed by the task's handler.
3795
>
validationFn: makeValidationFn(registrableTask, validate, chasmTask.Attempt, taskAttributes, taskValue),
3796
>
}
3797
>
3798
>
ctx = newContextWithOperationIntent(ctx, OperationIntentProgress)
3799
>
3800
>
defer log.CapturePanic(n.logger, &retErr)
3801
>
3802
>
return taskFn(ctx, ref, taskAttributes, taskValue.Interface())
3803
}
3804
3805
func (n *Node) ComponentByPath(
3806
chasmContext Context,
3807
path []string,
3809
>
node, ok := n.findNode(path)
3810
>
if !ok {
3811
return nil, errComponentNotFound
3812
}
3813
3815
return nil, err
3816
}
3817
3819
>
if !ok {
3820
return nil, softassert.UnexpectedInternalErr(
3821
n.logger,
3822
"component value is not of type Component",
3823
fmt.Errorf("%s", reflect.TypeOf(node.value).String()))
3824
}
3825
3827
}
3828
3829
// makeValidationFn adapts the TaskValidator interface to the ComponentRef's
3830
// validation callback format. Returns a validation function that wraps the
3831
// given validation callback to be called before the RegistrableTask's registered
3832
// validator callback. Intended for use to validate mutable state at access time.
3833
func makeValidationFn(
3834
registrableTask *RegistrableTask,
3835
validate func(NodeBackend, Context, Component) error,
3836
attempt int,
3837
taskAttributes TaskAttributes,
3838
taskValue reflect.Value,
3840
>
return func(backend NodeBackend, ctx Context, component Component, registry *Registry) error {
3841
>
// Call the provided validation callback.
3842
>
err := validate(backend, ctx, component)
3843
>
if err != nil {
3844
return err
3845
}
3846
3847
// Side effect's task validator is invoked inside the task handler,
3848
// so the panic wrapper ExecuteSideEffectTask() will cover this case.
3849
3850
// Call the TaskValidator.
3852
>
ctx,
3853
>
component,
3854
>
TaskInvocation{TaskAttributes: taskAttributes, Attempt: attempt},
3855
>
taskValue.Interface(),
3856
>
registry,
3857
>
)
3858
>
if err != nil {
3860
>
}
3863
>
}
3865
}
3866
}
3867
3868
// encodeChasmBlob encodes CHASM data and task payloads through the env-aware
3869
// serializer while preserving deterministic proto3 bytes for byte comparisons.
3871
>
return serialization.Encode(m, serialization.WithDeterministicProto3)
3872
>
}