go.temporal.io/server/chasm/tree_test.go
5299 LOC · 0 covered · 5299 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package chasm
2
3
import (
4
"context"
5
"errors"
6
"fmt"
7
"maps"
8
"reflect"
9
"slices"
10
"sort"
11
"strings"
12
"testing"
13
"time"
14
15
"github.com/stretchr/testify/require"
16
"github.com/stretchr/testify/suite"
17
commonpb "go.temporal.io/api/common/v1"
18
enumspb "go.temporal.io/api/enums/v1"
19
sdkpb "go.temporal.io/api/sdk/v1"
20
"go.temporal.io/api/serviceerror"
21
enumsspb "go.temporal.io/server/api/enums/v1"
22
persistencespb "go.temporal.io/server/api/persistence/v1"
23
"go.temporal.io/server/common"
24
"go.temporal.io/server/common/clock"
25
"go.temporal.io/server/common/definition"
26
"go.temporal.io/server/common/log"
27
"go.temporal.io/server/common/metrics"
28
"go.temporal.io/server/common/primitives"
29
"go.temporal.io/server/common/testing/protoassert"
30
"go.temporal.io/server/common/testing/protorequire"
31
"go.temporal.io/server/common/testing/testlogger"
32
"go.temporal.io/server/service/history/tasks"
33
"go.uber.org/mock/gomock"
34
"google.golang.org/protobuf/proto"
35
"google.golang.org/protobuf/types/known/timestamppb"
36
)
37
38
type (
39
nodeSuite struct {
40
suite.Suite
41
*require.Assertions
42
protorequire.ProtoAssertions
43
44
controller *gomock.Controller
45
nodeBackend *MockNodeBackend
46
testLibrary *TestLibrary
47
48
registry *Registry
49
timeSource *clock.EventTimeSource
50
nodePathEncoder NodePathEncoder
51
logger log.Logger
52
metricsHandler metrics.Handler
53
}
54
)
55
56
func TestNodeSuite(t *testing.T) {
57
suite.Run(t, new(nodeSuite))
58
}
59
60
func (s *nodeSuite) SetupTest() {
61
s.initAssertions()
62
s.controller = gomock.NewController(s.T())
63
s.nodeBackend = &MockNodeBackend{}
64
s.testLibrary = newTestLibrary(s.controller)
65
66
s.logger = testlogger.NewTestLogger(s.T(), testlogger.FailOnAnyUnexpectedError)
67
s.metricsHandler = metrics.NoopMetricsHandler
68
s.registry = NewRegistry(s.logger)
69
err := s.registry.Register(s.testLibrary)
70
s.NoError(err)
71
err = s.registry.Register(&CoreLibrary{})
72
s.NoError(err)
73
74
s.timeSource = clock.NewEventTimeSource()
75
s.nodePathEncoder = &testNodePathEncoder{}
76
}
77
78
func (s *nodeSuite) SetupSubTest() {
79
s.initAssertions()
80
}
81
82
func (s *nodeSuite) initAssertions() {
83
// `s.Assertions` (as well as other test helpers which depends on `s.T()`) must be initialized on
84
// both test and subtest levels (but not suite level, where `s.T()` is `nil`).
85
//
86
// If these helpers are not reinitialized on subtest level, any failed `assert` in
87
// subtest will fail the entire test (not subtest) immediately without running other subtests.
88
89
s.Assertions = require.New(s.T())
90
s.ProtoAssertions = protorequire.New(s.T())
91
}
92
93
func (s *nodeSuite) TestNewTree() {
94
persistenceNodes := map[string]*persistencespb.ChasmNode{
95
"": {
96
Metadata: &persistencespb.ChasmNodeMetadata{
97
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
98
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
99
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
100
TypeId: testComponentTypeID,
101
},
102
},
103
},
104
},
105
"child1": {
106
Metadata: &persistencespb.ChasmNodeMetadata{
107
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
108
},
109
},
110
"child2": {
111
Metadata: &persistencespb.ChasmNodeMetadata{
112
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
113
},
114
},
115
"child1/grandchild1": {
116
Metadata: &persistencespb.ChasmNodeMetadata{
117
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 4},
118
},
119
},
120
"child2/grandchild1": {
121
Metadata: &persistencespb.ChasmNodeMetadata{
122
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 5},
123
},
124
},
125
}
126
expectedPreorderNodes := []*persistencespb.ChasmNode{
127
persistenceNodes[""],
128
persistenceNodes["child1"],
129
persistenceNodes["child1/grandchild1"],
130
persistenceNodes["child2"],
131
persistenceNodes["child2/grandchild1"],
132
}
133
134
root, err := s.newTestTree(persistenceNodes)
135
s.NoError(err)
136
s.NotNil(root)
137
138
preorderNodes := s.preorderAndAssertParent(root, nil)
139
s.Len(preorderNodes, 5)
140
s.Equal(expectedPreorderNodes, preorderNodes)
141
}
142
143
func (s *nodeSuite) TestInitSerializedNode_TypeComponent() {
144
node := newNode(s.nodeBase(), nil, "")
145
node.initSerializedNode(fieldTypeComponent)
146
147
s.NotNil(node.serializedNode.GetMetadata().GetComponentAttributes(), "node serializedNode must have attributes created")
148
s.Nil(node.serializedNode.GetData(), "node serializedNode must not have data before serialize is called")
149
}
150
151
func (s *nodeSuite) TestSerializeNode_ComponentAttributes() {
152
node := s.testComponentTree()
153
154
s.Len(node.children, 2)
155
s.NotNil(node.children["SubComponent1"].value)
156
s.Len(node.children["SubComponent1"].children, 2)
157
s.NotNil(node.children["SubComponent1"].children["SubComponent11"].value)
158
s.Empty(node.children["SubComponent1"].children["SubComponent11"].children)
159
160
// Serialize root component.
161
s.NotNil(node.serializedNode.GetMetadata().GetComponentAttributes())
162
s.Nil(node.serializedNode.GetData())
163
err := node.serialize()
164
s.NoError(err)
165
s.NotNil(node.serializedNode)
166
s.NotNil(node.serializedNode.GetData(), "node serialized value must have data after serialize is called")
167
s.Equal(testComponentTypeID, node.serializedNode.GetMetadata().GetComponentAttributes().GetTypeId(), "node serialized value must have type set")
168
s.Equal(valueStateSynced, node.valueState)
169
170
// Serialize subcomponents (there are 2 subcomponents).
171
sc1Node := node.children["SubComponent1"]
172
s.NotNil(sc1Node.serializedNode.GetMetadata().GetComponentAttributes())
173
s.Nil(sc1Node.serializedNode.GetData())
174
for _, childNode := range node.children {
175
err = childNode.serialize()
176
s.NoError(err)
177
s.Equal(valueStateSynced, childNode.valueState)
178
}
179
s.NotNil(sc1Node.serializedNode.GetData(), "child node serialized value must have data after serialize is called")
180
s.Equal(testSubComponent1TypeID, sc1Node.serializedNode.GetMetadata().GetComponentAttributes().GetTypeId(), "node serialized value must have type set")
181
182
// Check SubData too.
183
sd1Node := node.children["SubData1"]
184
s.NoError(err)
185
s.NotNil(sd1Node.serializedNode.GetData(), "child node serialized value must have data after serialize is called")
186
}
187
188
func (s *nodeSuite) TestSerializeNode_ClearComponentData() {
189
node := s.testComponentTree()
190
191
node.value.(*TestComponent).ComponentData = nil
192
193
err := node.serialize()
194
s.NoError(err)
195
s.NotNil(node.serializedNode, "node serialized value must be not nil after serialize is called")
196
s.NotNil(node.serializedNode.GetMetadata().GetComponentAttributes(), "metadata must have component attributes")
197
s.Nil(node.serializedNode.GetData(), "data field must cleared to nil")
198
s.Equal(testComponentTypeID, node.serializedNode.GetMetadata().GetComponentAttributes().GetTypeId(), "type must present")
199
s.Equal(valueStateSynced, node.valueState)
200
}
201
202
func (s *nodeSuite) TestSerializeNode_ClearSubDataField() {
203
node := s.testComponentTree()
204
205
mutableContext := NewMutableContext(context.Background(), node)
206
component, err := node.Component(mutableContext, ComponentRef{})
207
s.NoError(err)
208
testComponent := component.(*TestComponent)
209
210
testComponent.SubData1 = NewEmptyField[*protoMessageType]()
211
212
sd1Node := node.children["SubData1"]
213
s.NotNil(sd1Node)
214
215
err = node.syncSubComponents()
216
s.NoError(err)
217
s.False(node.needsPointerResolution)
218
// SubData1 was never persisted (nil LVT), so no storage delete is needed.
219
s.Empty(node.mutation.DeletedNodes)
220
221
sd1Node = node.children["SubData1"]
222
s.Nil(sd1Node)
223
}
224
225
func (s *nodeSuite) TestSetRootComponent_SetsArchetypeID() {
226
rootNode := NewEmptyTree(s.registry, s.timeSource, s.nodeBackend, s.nodePathEncoder, s.logger, s.metricsHandler)
227
s.Equal(WorkflowArchetypeID, rootNode.ArchetypeID())
228
rootComponent := &TestComponent{
229
MSPointer: NewMSPointer(s.nodeBackend),
230
}
231
s.NoError(rootNode.SetRootComponent(rootComponent))
232
s.Equal(testComponentTypeID, rootNode.ArchetypeID())
233
s.NotEqual(WorkflowArchetypeID, rootNode.ArchetypeID())
234
}
235
236
func (s *nodeSuite) TestInitSerializedNode_TypeData() {
237
node := newNode(s.nodeBase(), nil, "")
238
node.initSerializedNode(fieldTypeData)
239
s.NotNil(node.serializedNode.GetMetadata().GetDataAttributes(), "node serializedNode must have attributes created")
240
s.Nil(node.serializedNode.GetData(), "node serializedNode must not have data before serialize is called")
241
}
242
243
func (s *nodeSuite) TestSerializeNode_DataAttributes() {
244
component := &protoMessageType{
245
CreateRequestId: "22",
246
}
247
248
node := newNode(s.nodeBase(), nil, "")
249
node.initSerializedNode(fieldTypeData)
250
node.value = component
251
node.valueState = valueStateNeedSerialize
252
253
err := node.serialize()
254
s.NoError(err)
255
s.NotNil(node.serializedNode.GetData(), "child node serialized value must have data after serialize is called")
256
s.Equal(enumspb.ENCODING_TYPE_PROTO3, node.serializedNode.GetData().GetEncodingType())
257
s.Equal([]byte{0xa, 0x2, 0x32, 0x32}, node.serializedNode.GetData().GetData())
258
s.Equal(valueStateSynced, node.valueState)
259
}
260
261
func (s *nodeSuite) TestCollectionAttributes() {
262
runID1 := fmt.Sprintf("workflow_id_%d", 1)
263
runID2 := fmt.Sprintf("workflow_id_%d", 2)
264
sc1 := &TestSubComponent1{
265
SubComponent1Data: &protoMessageType{
266
RunId: runID1,
267
},
268
}
269
sc2 := &TestSubComponent1{
270
SubComponent1Data: &protoMessageType{
271
RunId: runID2,
272
},
273
}
274
275
type testCase struct {
276
name string
277
initComponent func() *TestComponent
278
mapField string
279
}
280
cases := []testCase{
281
{
282
name: "of string key",
283
initComponent: func() *TestComponent {
284
return &TestComponent{
285
SubComponents: Map[string, *TestSubComponent1]{
286
"SubComponent1": NewComponentField(nil, sc1),
287
"SubComponent2": NewComponentField(nil, sc2),
288
},
289
}
290
},
291
mapField: "SubComponents",
292
},
293
{
294
name: "of int key",
295
initComponent: func() *TestComponent {
296
return &TestComponent{
297
PendingActivities: Map[int, *TestSubComponent1]{
298
1: NewComponentField(nil, sc1),
299
2: NewComponentField(nil, sc2),
300
},
301
}
302
},
303
mapField: "PendingActivities",
304
},
305
}
306
307
for _, tc := range cases {
308
309
var persistedNodes map[string]*persistencespb.ChasmNode
310
311
s.Run("Sync and serialize component with map "+tc.name, func() {
312
var nilSerializedNodes map[string]*persistencespb.ChasmNode
313
rootNode, err := s.newTestTree(nilSerializedNodes)
314
s.NoError(err)
315
316
rootComponent := tc.initComponent()
317
err = rootNode.SetRootComponent(rootComponent)
318
s.NoError(err)
319
320
mutations, err := rootNode.CloseTransaction()
321
s.NoError(err)
322
s.Len(mutations.UpdatedNodes, 4, "root, collection, and 2 collection items must be updated")
323
s.Empty(mutations.DeletedNodes)
324
325
switch tc.mapField {
326
case "SubComponents":
327
s.NotEmpty(rootNode.children[tc.mapField].children["SubComponent1"].serializedNode.GetData().GetData())
328
s.NotEmpty(rootNode.children[tc.mapField].children["SubComponent2"].serializedNode.GetData().GetData())
329
case "PendingActivities":
330
s.NotEmpty(rootNode.children[tc.mapField].children["1"].serializedNode.GetData().GetData())
331
s.NotEmpty(rootNode.children[tc.mapField].children["2"].serializedNode.GetData().GetData())
332
}
333
334
// Save it use in other subtests.
335
persistedNodes = common.CloneProtoMap(mutations.UpdatedNodes)
336
})
337
338
s.NotNil(persistedNodes)
339
340
s.Run("Deserialize component with map "+tc.name, func() {
341
rootNode, err := s.newTestTree(persistedNodes)
342
s.NoError(err)
343
344
err = rootNode.deserialize(reflect.TypeFor[*TestComponent]())
345
s.NoError(err)
346
347
rootComponent := rootNode.value.(*TestComponent)
348
349
var sc1Field, sc2Field Field[*TestSubComponent1]
350
switch tc.mapField {
351
case "SubComponents":
352
s.NotNil(rootComponent.SubComponents)
353
s.Len(rootComponent.SubComponents, 2)
354
sc1Field, sc2Field = rootComponent.SubComponents["SubComponent1"], rootComponent.SubComponents["SubComponent2"]
355
case "PendingActivities":
356
s.NotNil(rootComponent.PendingActivities)
357
s.Len(rootComponent.PendingActivities, 2)
358
sc1Field, sc2Field = rootComponent.PendingActivities[1], rootComponent.PendingActivities[2]
359
}
360
361
chasmContext := NewMutableContext(context.Background(), rootNode)
362
sc1Des := sc1Field.Get(chasmContext)
363
s.Equal(sc1.SubComponent1Data.GetRunId(), sc1Des.SubComponent1Data.GetRunId())
364
365
sc2Des := sc2Field.Get(chasmContext)
366
s.Equal(sc2.SubComponent1Data.GetRunId(), sc2Des.SubComponent1Data.GetRunId())
367
})
368
369
s.Run("Clear map "+tc.name+" by setting it to nil", func() {
370
rootNode, err := s.newTestTree(persistedNodes)
371
s.NoError(err)
372
373
err = rootNode.deserialize(reflect.TypeFor[*TestComponent]())
374
s.NoError(err)
375
376
rootComponent := rootNode.value.(*TestComponent)
377
378
rootNode.valueState = valueStateNeedSyncStructure
379
switch tc.mapField {
380
case "SubComponents":
381
rootComponent.SubComponents = nil
382
case "PendingActivities":
383
rootComponent.PendingActivities = nil
384
}
385
386
mutation, err := rootNode.CloseTransaction()
387
s.NoError(err)
388
s.Empty(mutation.UpdatedNodes, "root component data is unchanged; collection deletion is tracked by DeletedNodes")
389
s.Len(mutation.DeletedNodes, 3, "collection and 2 collection items must be deleted")
390
})
391
392
s.Run("Delete single map "+tc.name+" item", func() {
393
rootNode, err := s.newTestTree(persistedNodes)
394
s.NoError(err)
395
396
err = rootNode.deserialize(reflect.TypeFor[*TestComponent]())
397
s.NoError(err)
398
399
rootComponent := rootNode.value.(*TestComponent)
400
401
// Delete collection item 1.
402
rootNode.valueState = valueStateNeedSyncStructure
403
switch tc.mapField {
404
case "SubComponents":
405
delete(rootComponent.SubComponents, "SubComponent1")
406
case "PendingActivities":
407
delete(rootComponent.PendingActivities, 1)
408
}
409
410
mutation, err := rootNode.CloseTransaction()
411
s.NoError(err)
412
s.Empty(mutation.UpdatedNodes, "root component data is unchanged; collection item deletion is tracked by DeletedNodes")
413
s.Len(mutation.DeletedNodes, 1, "collection item 1 must be deleted")
414
})
415
416
s.Run("Clear map "+tc.name+" by deleting all items", func() {
417
rootNode, err := s.newTestTree(persistedNodes)
418
s.NoError(err)
419
420
err = rootNode.deserialize(reflect.TypeFor[*TestComponent]())
421
s.NoError(err)
422
423
rootComponent := rootNode.value.(*TestComponent)
424
425
// Delete both collection items.
426
rootNode.valueState = valueStateNeedSyncStructure
427
switch tc.mapField {
428
case "SubComponents":
429
delete(rootComponent.SubComponents, "SubComponent1")
430
delete(rootComponent.SubComponents, "SubComponent2")
431
case "PendingActivities":
432
delete(rootComponent.PendingActivities, 1)
433
delete(rootComponent.PendingActivities, 2)
434
}
435
436
// Now map is empty and must be deleted.
437
mutation, err := rootNode.CloseTransaction()
438
s.NoError(err)
439
s.Empty(mutation.UpdatedNodes, "root component data is unchanged; collection deletion is tracked by DeletedNodes")
440
s.Len(mutation.DeletedNodes, 3, "collection and 2 items must be deleted")
441
})
442
443
s.Run("Nil map "+tc.name+" on first transaction produces no deletions", func() {
444
// A map field that was never set (nil) should not produce any DeletedNodes
445
// entries when the first transaction is closed — there is nothing in persistence
446
// to delete.
447
var nilSerializedNodes map[string]*persistencespb.ChasmNode
448
rootNode, err := s.newTestTree(nilSerializedNodes)
449
s.NoError(err)
450
451
err = rootNode.SetRootComponent(&TestComponent{}) // all map fields are nil
452
s.NoError(err)
453
454
mutation, err := rootNode.CloseTransaction()
455
s.NoError(err)
456
s.Empty(mutation.DeletedNodes, "no nodes should be deleted for a map that never existed")
457
})
458
459
s.Run("Empty (non-nil) map "+tc.name+" on first transaction produces no deletions", func() {
460
// A map field initialized to an empty (non-nil) map should also not produce
461
// any DeletedNodes entries — an empty map is equivalent to nil at the
462
// persistence layer and there is nothing to delete.
463
var nilSerializedNodes map[string]*persistencespb.ChasmNode
464
rootNode, err := s.newTestTree(nilSerializedNodes)
465
s.NoError(err)
466
467
var rootComponent TestComponent
468
switch tc.mapField {
469
case "SubComponents":
470
rootComponent.SubComponents = Map[string, *TestSubComponent1]{}
471
case "PendingActivities":
472
rootComponent.PendingActivities = Map[int, *TestSubComponent1]{}
473
default:
474
s.Failf("unexpected mapField", "unknown mapField %q in test case", tc.mapField)
475
}
476
err = rootNode.SetRootComponent(&rootComponent)
477
s.NoError(err)
478
479
mutation, err := rootNode.CloseTransaction()
480
s.NoError(err)
481
s.Empty(mutation.DeletedNodes, "no nodes should be deleted for a newly-created empty map")
482
})
483
}
484
}
485
486
func (s *nodeSuite) TestMapDeserializeNilToEmpty() {
487
// Verify that a Map field that was never set deserializes to an empty (non-nil)
488
// map so callers can range over it without nil checks.
489
var nilSerializedNodes map[string]*persistencespb.ChasmNode
490
rootNode, err := s.newTestTree(nilSerializedNodes)
491
s.NoError(err)
492
493
err = rootNode.SetRootComponent(&TestComponent{})
494
s.NoError(err)
495
496
mutations, err := rootNode.CloseTransaction()
497
s.NoError(err)
498
// Only root is updated; no collection nodes because maps were nil/empty.
499
s.Len(mutations.UpdatedNodes, 1)
500
s.Empty(mutations.DeletedNodes)
501
502
persistedNodes := common.CloneProtoMap(mutations.UpdatedNodes)
503
504
rootNode2, err := s.newTestTree(persistedNodes)
505
s.NoError(err)
506
507
err = rootNode2.deserialize(reflect.TypeFor[*TestComponent]())
508
s.NoError(err)
509
510
rootComponent := rootNode2.value.(*TestComponent)
511
s.NotNil(rootComponent.SubComponents, "SubComponents must be non-nil after deserialization")
512
s.Empty(rootComponent.SubComponents)
513
s.NotNil(rootComponent.PendingActivities, "PendingActivities must be non-nil after deserialization")
514
s.Empty(rootComponent.PendingActivities)
515
}
516
517
func (s *nodeSuite) TestPointerAttributes() {
518
var persistedNodes map[string]*persistencespb.ChasmNode
519
520
sc11 := &TestSubComponent11{
521
SubComponent11Data: &protoMessageType{
522
RunId: fmt.Sprintf("workflow_id_%d", 11),
523
},
524
}
525
526
sc1 := &TestSubComponent1{
527
SubComponent1Data: &protoMessageType{
528
RunId: fmt.Sprintf("workflow_id_%d", 1),
529
},
530
SubComponent11: NewComponentField(nil, sc11),
531
}
532
533
s.Run("Sync and serialize component with ancestor pointer", func() {
534
var nilSerializedNodes map[string]*persistencespb.ChasmNode
535
rootNode, err := s.newTestTree(nilSerializedNodes)
536
s.NoError(err)
537
538
ctx := NewMutableContext(context.Background(), rootNode)
539
540
rootComponent := &TestComponent{
541
MSPointer: NewMSPointer(s.nodeBackend),
542
SubComponent1: NewComponentField(nil, sc1),
543
SubComponentInterfacePointer: NewComponentField[Component](nil, sc1),
544
}
545
546
// sc11 points to root (grandparent) -- an ancestor pointer.
547
sc11.GrandparentPointer = ComponentPointerTo(ctx, rootComponent)
548
549
s.NoError(rootNode.SetRootComponent(rootComponent))
550
551
s.Equal(fieldTypeDeferredPointer, sc11.GrandparentPointer.Internal.ft)
552
553
mutations, err := rootNode.CloseTransaction()
554
s.NoError(err)
555
s.Len(mutations.UpdatedNodes, 5, "root, SubComponent1, SubComponent11, GrandparentPointer, and SubComponentInterfacePointer must be updated")
556
s.Empty(mutations.DeletedNodes)
557
558
sc11Node := rootNode.children["SubComponent1"].children["SubComponent11"]
559
s.Equal(
560
[]string{},
561
sc11Node.children["GrandparentPointer"].serializedNode.GetMetadata().GetPointerAttributes().GetNodePath(),
562
)
563
564
// Save for use in other subtests.
565
persistedNodes = common.CloneProtoMap(mutations.UpdatedNodes)
566
})
567
568
s.NotNil(persistedNodes)
569
570
s.Run("Deserialize ancestor pointer component", func() {
571
rootNode, err := s.newTestTree(persistedNodes)
572
s.NoError(err)
573
574
mutableContext := NewMutableContext(context.Background(), rootNode)
575
component, err := rootNode.Component(mutableContext, ComponentRef{})
576
s.NoError(err)
577
testComponent := component.(*TestComponent)
578
579
s.NotNil(testComponent.MSPointer)
580
581
chasmContext := NewMutableContext(context.Background(), rootNode)
582
sc1Des := testComponent.SubComponent1.Get(chasmContext)
583
s.NotNil(sc1Des)
584
sc11Des := sc1Des.SubComponent11.Get(chasmContext)
585
s.NotNil(sc11Des)
586
587
rootViaPointer := sc11Des.GrandparentPointer.Get(chasmContext)
588
s.NotNil(rootViaPointer)
589
s.Equal(testComponent, rootViaPointer)
590
591
ifacePtr := testComponent.SubComponentInterfacePointer.Get(chasmContext)
592
s.NotNil(ifacePtr)
593
594
sc1ptr, ok := ifacePtr.(*TestSubComponent1)
595
s.True(ok)
596
s.ProtoEqual(sc1ptr.SubComponent1Data, sc1.SubComponent1Data)
597
})
598
599
s.Run("Clear ancestor pointer by setting it to the empty field", func() {
600
rootNode, err := s.newTestTree(persistedNodes)
601
s.NoError(err)
602
603
mutableContext := NewMutableContext(context.Background(), rootNode)
604
component, err := rootNode.Component(mutableContext, ComponentRef{})
605
s.NoError(err)
606
testComponent := component.(*TestComponent)
607
sc1Des := testComponent.SubComponent1.Get(mutableContext)
608
sc11Des := sc1Des.SubComponent11.Get(mutableContext)
609
610
sc11Des.GrandparentPointer = NewEmptyField[*TestComponent]()
611
612
mutation, err := rootNode.CloseTransaction()
613
s.NoError(err)
614
s.Empty(mutation.UpdatedNodes)
615
s.Len(mutation.DeletedNodes, 1, "GrandparentPointer must be deleted")
616
})
617
}
618
619
func (s *nodeSuite) TestParentPointer_InMemory() {
620
node := s.testComponentTree()
621
622
s.assertParentPointer(node)
623
624
// Additionally also test parentPtr for components inside a map.
625
626
mutableContext := NewMutableContext(context.Background(), node)
627
component, err := node.Component(mutableContext, ComponentRef{})
628
s.NoError(err)
629
testComponent := component.(*TestComponent)
630
631
mapSubComponent1 := &TestSubComponent1{}
632
// Try using the testComponent we get from the ParentPtr for the mutation.
633
testComponent.SubComponents = Map[string, *TestSubComponent1]{
634
"mapSubComponent1": NewComponentField(mutableContext, mapSubComponent1),
635
}
636
637
s.Panics(func() {
638
_ = mapSubComponent1.ParentPtr.Get(mutableContext)
639
})
640
641
// Sync structure initializes the parent pointer
642
err = node.syncSubComponents()
643
s.NoError(err)
644
645
testComponentFromPtr := mapSubComponent1.ParentPtr.Get(mutableContext)
646
// Asserting they actually point to the same testComponent object.
647
s.Same(testComponent, testComponentFromPtr)
648
}
649
650
func (s *nodeSuite) TestParentPointer_FromDB() {
651
serializedNodes := testComponentSerializedNodes()
652
653
node, err := s.newTestTree(serializedNodes)
654
s.NoError(err)
655
656
s.assertParentPointer(node)
657
}
658
659
func (s *nodeSuite) assertParentPointer(testComponentNode *Node) {
660
chasmContext := NewContext(context.Background(), testComponentNode)
661
component, err := testComponentNode.Component(chasmContext, ComponentRef{})
662
s.NoError(err)
663
testComponent := component.(*TestComponent)
664
665
_, found := testComponent.ParentPtr.TryGet(chasmContext)
666
s.False(found)
667
668
subComponent1 := testComponent.SubComponent1.Get(chasmContext)
669
testComponentFromPtr := subComponent1.ParentPtr.Get(chasmContext)
670
// Asserting they actually point to the same testComponent object.
671
s.Same(testComponent, testComponentFromPtr)
672
673
subComponent11 := subComponent1.SubComponent11.Get(chasmContext)
674
testSubComponent1FromPtr := subComponent11.ParentPtr.Get(chasmContext)
675
// Asserting they actually point to the same testSubComponent1 object.
676
s.Same(subComponent1, testSubComponent1FromPtr)
677
}
678
679
func (s *nodeSuite) TestSyncSubComponents_DeleteLeafNode() {
680
node := s.testComponentTree()
681
682
mutableContext := NewMutableContext(context.Background(), node)
683
component, err := node.ComponentByPath(mutableContext, []string{"SubComponent1"})
684
s.NoError(err)
685
686
sc1 := component.(*TestSubComponent1)
687
sc1.SubComponent11 = NewEmptyField[*TestSubComponent11]()
688
s.NotNil(node.children["SubComponent1"].children["SubComponent11"])
689
690
err = node.syncSubComponents()
691
s.NoError(err)
692
s.False(node.needsPointerResolution)
693
694
// SubComponent11 was never persisted (nil LVT), so no storage delete is needed.
695
s.Empty(node.mutation.DeletedNodes)
696
s.Nil(node.children["SubComponent1"].children["SubComponent11"])
697
}
698
699
func (s *nodeSuite) TestSyncSubComponents_DeleteMiddleNode() {
700
node := s.testComponentTree()
701
702
mutableContext := NewMutableContext(context.Background(), node)
703
component, err := node.Component(mutableContext, ComponentRef{})
704
s.NoError(err)
705
testComponent := component.(*TestComponent)
706
707
// Set subcomponent at middle node to nil.
708
testComponent.SubComponent1 = NewEmptyField[*TestSubComponent1]()
709
s.NotNil(node.children["SubComponent1"])
710
711
err = node.syncSubComponents()
712
s.NoError(err)
713
s.False(node.needsPointerResolution)
714
715
// SubComponent1 and its children were never persisted (nil LVT), so no storage deletes are needed.
716
s.Empty(node.mutation.DeletedNodes)
717
718
s.Nil(node.children["SubComponent1"])
719
}
720
721
func (s *nodeSuite) TestDeserializeNode_EmptyPersistence() {
722
var serializedNodes map[string]*persistencespb.ChasmNode
723
724
node, err := s.newTestTree(serializedNodes)
725
s.NoError(err)
726
s.Nil(node.value)
727
s.NotNil(node.serializedNode)
728
729
err = node.deserialize(reflect.TypeFor[*TestComponent]())
730
s.NoError(err)
731
s.NotNil(node.value)
732
s.IsType(&TestComponent{}, node.value)
733
tc := node.value.(*TestComponent)
734
s.Equal(valueStateSynced, node.valueState)
735
s.Nil(tc.SubComponent1.Internal.node)
736
s.Nil(tc.SubComponent1.Internal.value())
737
738
// nil component data should decode into zero value
739
s.NotNil(tc.ComponentData)
740
s.ProtoEqual(&protoMessageType{}, tc.ComponentData)
741
}
742
743
func (s *nodeSuite) TestDeserializeNode_ComponentAttributes() {
744
serializedNodes := testComponentSerializedNodes()
745
746
// Root component will be deserialized as part of the initialization process,
747
// for initializing search attributes and memo.
748
node, err := s.newTestTree(serializedNodes)
749
s.NoError(err)
750
s.NotNil(node.serializedNode)
751
s.NotNil(node.value)
752
s.IsType(&TestComponent{}, node.value)
753
tc := node.value.(*TestComponent)
754
s.Equal(tc.SubComponent1.Internal.node, node.children["SubComponent1"])
755
s.Equal("component-data", tc.ComponentData.CreateRequestId)
756
s.Equal(valueStateSynced, node.valueState)
757
758
s.Nil(tc.SubComponent1.Internal.value())
759
s.Equal(valueStateNeedDeserialize, tc.SubComponent1.Internal.node.valueState)
760
err = tc.SubComponent1.Internal.node.deserialize(reflect.TypeFor[*TestSubComponent1]())
761
s.NoError(err)
762
s.NotNil(tc.SubComponent1.Internal.node.value)
763
s.IsType(&TestSubComponent1{}, tc.SubComponent1.Internal.node.value)
764
s.Equal("sub-component1-data", tc.SubComponent1.Internal.node.value.(*TestSubComponent1).SubComponent1Data.CreateRequestId)
765
s.Equal(valueStateSynced, tc.SubComponent1.Internal.node.valueState)
766
}
767
768
func (s *nodeSuite) TestDeserializeNode_DataAttributes() {
769
serializedNodes := testComponentSerializedNodes()
770
771
// Root component will be deserialized as part of the initialization process,
772
// for initializing search attributes and memo.
773
node, err := s.newTestTree(serializedNodes)
774
s.NoError(err)
775
s.NotNil(node.serializedNode)
776
s.NotNil(node.value)
777
s.Equal(valueStateSynced, node.valueState)
778
779
s.IsType(&TestComponent{}, node.value)
780
tc := node.value.(*TestComponent)
781
782
s.Equal(tc.SubData1.Internal.node, node.children["SubData1"])
783
784
s.Nil(tc.SubData1.Internal.value())
785
err = tc.SubData1.Internal.node.deserialize(reflect.TypeFor[*protoMessageType]())
786
s.NoError(err)
787
s.NotNil(tc.SubData1.Internal.node.value)
788
s.Equal(valueStateSynced, tc.SubData1.Internal.node.valueState)
789
s.IsType(&protoMessageType{}, tc.SubData1.Internal.node.value)
790
s.Equal("sub-data1", tc.SubData1.Internal.node.value.(*protoMessageType).CreateRequestId)
791
}
792
793
func (s *nodeSuite) TestFieldInterface() {
794
type testComponent struct {
795
UnimplementedComponent
796
Data *protoMessageType
797
SubComponent1 Field[TestSubComponent]
798
}
799
800
serializedNodes := testComponentSerializedNodes()
801
node, err := s.newTestTree(serializedNodes)
802
s.NoError(err)
803
804
err = node.deserialize(reflect.TypeFor[*testComponent]())
805
s.NoError(err)
806
s.NotNil(node.value)
807
s.IsType(&testComponent{}, node.value)
808
tc := node.value.(*testComponent)
809
810
chasmContext := NewMutableContext(context.Background(), node)
811
sc1 := tc.SubComponent1.Get(chasmContext)
812
s.NotNil(sc1)
813
s.Equal("sub-component1-data", sc1.GetData())
814
}
815
816
func (s *nodeSuite) TestGenerateSerializedNodes() {
817
s.T().Skip("This test is used to generate serialized nodes for other tests.")
818
819
node := s.testComponentTree()
820
821
err := node.serialize()
822
s.NoError(err)
823
serializedNodes := map[string]*persistencespb.ChasmNode{}
824
serializedNodes[""] = node.serializedNode
825
826
for childName, childNode := range node.children {
827
err = childNode.serialize()
828
s.NoError(err)
829
serializedNodes[childName] = childNode.serializedNode
830
}
831
832
for childName, childNode := range node.children["SubComponent1"].children {
833
err = childNode.serialize()
834
s.NoError(err)
835
serializedNodes["SubComponent1/"+childName] = childNode.serializedNode
836
}
837
838
generateMapInit(serializedNodes, "serializedNodes")
839
}
840
841
func (s *nodeSuite) TestNodeSnapshot() {
842
persistenceNodes := map[string]*persistencespb.ChasmNode{
843
"": {
844
Metadata: &persistencespb.ChasmNodeMetadata{
845
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
846
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
847
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
848
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
849
TypeId: testComponentTypeID,
850
},
851
},
852
},
853
},
854
"child1": {
855
Metadata: &persistencespb.ChasmNodeMetadata{
856
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 4},
857
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 4},
858
},
859
},
860
"child2": {
861
Metadata: &persistencespb.ChasmNodeMetadata{
862
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
863
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
864
},
865
},
866
"child1/grandchild1": {
867
Metadata: &persistencespb.ChasmNodeMetadata{
868
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
869
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
870
},
871
},
872
"child2/grandchild1": {
873
Metadata: &persistencespb.ChasmNodeMetadata{
874
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 5},
875
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 5},
876
},
877
},
878
}
879
880
root, err := s.newTestTree(persistenceNodes)
881
s.NoError(err)
882
s.NotNil(root)
883
884
// Test snapshot with nil exclusiveMinVT, which should return all nodes
885
snapshot := root.Snapshot(nil)
886
s.Equal(persistenceNodes, snapshot.Nodes)
887
888
// Test snapshot with non-nil exclusiveMinVT, which should return only nodes with higher
889
// LastUpdateVersionedTransition than the exclusiveMinVT
890
expectedNodePaths := []string{"child1", "child2/grandchild1"}
891
expectedNodes := make(map[string]*persistencespb.ChasmNode)
892
for _, path := range expectedNodePaths {
893
expectedNodes[path] = persistenceNodes[path]
894
}
895
snapshot = root.Snapshot(&persistencespb.VersionedTransition{TransitionCount: 3})
896
s.Equal(expectedNodes, snapshot.Nodes)
897
}
898
899
func (s *nodeSuite) TestApplyMutation() {
900
mustEncode := func(m proto.Message) *commonpb.DataBlob {
901
taskBlob, err := encodeChasmBlob(m)
902
s.NoError(err)
903
return taskBlob
904
}
905
906
now := s.timeSource.Now()
907
persistenceNodes := map[string]*persistencespb.ChasmNode{
908
"": {
909
Metadata: &persistencespb.ChasmNodeMetadata{
910
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
911
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
912
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
913
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
914
TypeId: testComponentTypeID,
915
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
916
{
917
// This task is not updated, so it's deserialized version will
918
// NOT be cleared below as part of the updateNode process.
919
TypeId: testPureTaskTypeID,
920
ScheduledTime: timestamppb.New(now.Add(time.Second)),
921
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
922
VersionedTransitionOffset: 1,
923
PhysicalTaskStatus: physicalTaskStatusNone,
924
Data: mustEncode(&commonpb.Payload{
925
Data: []byte("root-task-data-1"),
926
}),
927
},
928
{
929
// Task will be deleted, so deserialized version of this task should also be deleted from cache.
930
TypeId: testPureTaskTypeID,
931
ScheduledTime: timestamppb.New(now.Add(time.Second)),
932
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
933
VersionedTransitionOffset: 2,
934
PhysicalTaskStatus: physicalTaskStatusNone,
935
Data: mustEncode(&commonpb.Payload{
936
Data: []byte("root-task-data-2"),
937
}),
938
},
939
},
940
},
941
},
942
},
943
},
944
"SubComponent1": {
945
Metadata: &persistencespb.ChasmNodeMetadata{
946
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
947
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
948
},
949
},
950
"SubComponent1/SubComponent11": {
951
Metadata: &persistencespb.ChasmNodeMetadata{
952
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
953
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
954
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
955
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
956
TypeId: testSubComponent11TypeID,
957
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
958
{
959
// Node is deleted, so deserialized version of this task should be deleted from cache.
960
TypeId: testPureTaskTypeID,
961
ScheduledTime: timestamppb.New(now.Add(time.Minute)),
962
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
963
VersionedTransitionOffset: 3,
964
PhysicalTaskStatus: physicalTaskStatusNone,
965
Data: mustEncode(&commonpb.Payload{
966
Data: []byte("SubComponent11-task-data"),
967
}),
968
},
969
},
970
},
971
},
972
},
973
},
974
"SubComponent1/SubComponent11/SubComponent11Data": {
975
Metadata: &persistencespb.ChasmNodeMetadata{
976
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 4},
977
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 4},
978
},
979
},
980
}
981
root, err := s.newTestTree(persistenceNodes)
982
s.NoError(err)
983
s.Len(root.currentSA, 3)
984
s.NotNil(root.currentMemo)
985
initialMemo, ok := root.currentMemo.(*protoMessageType)
986
s.True(ok)
987
s.ProtoEqual(&protoMessageType{}, initialMemo)
988
989
// Manually deserialize some tasks to populate the taskValueCache
990
_, err = root.deserializeComponentTask(root.serializedNode.Metadata.GetComponentAttributes().PureTasks[0])
991
s.NoError(err)
992
_, err = root.deserializeComponentTask(root.serializedNode.Metadata.GetComponentAttributes().PureTasks[1])
993
s.NoError(err)
994
_, err = root.deserializeComponentTask(root.children["SubComponent1"].children["SubComponent11"].serializedNode.Metadata.GetComponentAttributes().PureTasks[0])
995
s.NoError(err)
996
s.Len(root.taskValueCache, 3)
997
998
// This decoded value should be reset after applying the mutation
999
root.children["SubComponent1"].value = "some-random-decoded-value"
1000
1001
// Prepare mutation: update root and "SubComponent1" node, delete "SubComponent1/SubComponent11", and add "newchild".
1002
1003
updatedRoot := &persistencespb.ChasmNode{
1004
Metadata: &persistencespb.ChasmNodeMetadata{
1005
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 30},
1006
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 30},
1007
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1008
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1009
TypeId: testComponentTypeID,
1010
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
1011
{
1012
TypeId: testPureTaskTypeID,
1013
ScheduledTime: timestamppb.New(now.Add(time.Second)),
1014
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1015
VersionedTransitionOffset: 1,
1016
PhysicalTaskStatus: physicalTaskStatusNone,
1017
Data: mustEncode(&commonpb.Payload{
1018
Data: []byte("root-task-data-1"),
1019
}),
1020
},
1021
},
1022
},
1023
},
1024
},
1025
Data: mustEncode(
1026
&protoMessageType{
1027
StartTime: timestamppb.New(now),
1028
}),
1029
}
1030
updatedSC1 := &persistencespb.ChasmNode{
1031
Metadata: &persistencespb.ChasmNodeMetadata{
1032
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 20},
1033
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 20},
1034
},
1035
}
1036
newSC2 := &persistencespb.ChasmNode{
1037
Metadata: &persistencespb.ChasmNodeMetadata{
1038
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 100},
1039
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 100},
1040
},
1041
}
1042
mutation := NodesMutation{
1043
UpdatedNodes: map[string]*persistencespb.ChasmNode{
1044
"": updatedRoot,
1045
"SubComponent1": updatedSC1,
1046
"SubComponent2": newSC2,
1047
},
1048
DeletedNodes: map[string]struct{}{
1049
"SubComponent1/SubComponent11": {}, // this should remove the entire "SubComponent11" subtree
1050
"SubComponent1/non-exist-child": {},
1051
},
1052
}
1053
err = root.ApplyMutation(mutation)
1054
s.NoError(err)
1055
1056
// Validate root node got updated.
1057
s.Equal(updatedRoot, root.serializedNode)
1058
s.NotNil(root.value)
1059
s.Len(root.currentSA, 3)
1060
s.Len(root.currentSA, 3)
1061
s.Contains(root.currentSA, "TemporalDatetime01")
1062
s.True(root.currentSA["TemporalDatetime01"].(VisibilityValueTime).Equal(VisibilityValueTime(now)))
1063
1064
// Validate memo content.
1065
s.NotNil(root.currentMemo)
1066
decodedMemo, ok := root.currentMemo.(*protoMessageType)
1067
s.True(ok, "currentMemo should be of type *protoMessageType")
1068
s.True(decodedMemo.StartTime.AsTime().Equal(now))
1069
1070
// Validate the "child" node got updated.
1071
nodeSC1, ok := root.children["SubComponent1"]
1072
s.True(ok)
1073
s.Equal(updatedSC1, nodeSC1.serializedNode)
1074
s.Nil(nodeSC1.value) // value should be reset after mutation
1075
1076
// Validate the "newchild" node is added.
1077
nodeSC2, ok := root.children["SubComponent2"]
1078
s.True(ok)
1079
s.Equal(newSC2, nodeSC2.serializedNode)
1080
1081
// Validate the "grandchild" node is deleted.
1082
s.Empty(nodeSC1.children)
1083
1084
// Validate that nodeBase.mutation reflects the applied mutation.
1085
// Only updates on existing nodes are recorded; new nodes are inserted without a mutation record.
1086
expectedMutation := NodesMutation{
1087
UpdatedNodes: map[string]*persistencespb.ChasmNode{
1088
"": updatedRoot,
1089
"SubComponent1": updatedSC1,
1090
"SubComponent2": newSC2,
1091
},
1092
DeletedNodes: map[string]struct{}{
1093
"SubComponent1/SubComponent11": {},
1094
"SubComponent1/SubComponent11/SubComponent11Data": {},
1095
},
1096
}
1097
s.Equal(expectedMutation, root.mutation)
1098
1099
s.Len(root.taskValueCache, 1)
1100
}
1101
1102
func (s *nodeSuite) TestApplyMutation_InvalidatesHydratedMapAncestors() {
1103
s.nodeBackend.HandleGetCurrentVersion = func() int64 { return 1 }
1104
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 1 }
1105
1106
newRootComponent := func(items map[string]string) *TestComponent {
1107
component := &TestComponent{
1108
ComponentData: &protoMessageType{
1109
RunId: "root",
1110
StartTime: timestamppb.New(s.timeSource.Now()),
1111
},
1112
SubComponents: make(Map[string, *TestSubComponent1], len(items)),
1113
}
1114
for key, runID := range items {
1115
component.SubComponents[key] = NewComponentField(nil, &TestSubComponent1{
1116
SubComponent1Data: &protoMessageType{RunId: runID},
1117
})
1118
}
1119
return component
1120
}
1121
1122
buildSnapshot := func(component *TestComponent) map[string]*persistencespb.ChasmNode {
1123
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 1 }
1124
root, err := s.newTestTree(nil)
1125
s.NoError(err)
1126
s.NoError(root.SetRootComponent(component))
1127
mutation, err := root.CloseTransaction()
1128
s.NoError(err)
1129
s.NotEmpty(mutation.UpdatedNodes)
1130
return common.CloneProtoMap(mutation.UpdatedNodes)
1131
}
1132
1133
mutationFromSource := func(
1134
persistedNodes map[string]*persistencespb.ChasmNode,
1135
mutate func(Context, *TestComponent),
1136
) NodesMutation {
1137
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
1138
source, err := s.newTestTree(common.CloneProtoMap(persistedNodes))
1139
s.NoError(err)
1140
chasmContext := NewMutableContext(context.Background(), source)
1141
component, err := source.Component(chasmContext, ComponentRef{})
1142
s.NoError(err)
1143
mutate(chasmContext, component.(*TestComponent))
1144
mutation, err := source.CloseTransaction()
1145
s.NoError(err)
1146
s.NotContains(mutation.UpdatedNodes, "", "replicated mutation must not include the hydrated parent component")
1147
return NodesMutation{
1148
UpdatedNodes: common.CloneProtoMap(mutation.UpdatedNodes),
1149
DeletedNodes: maps.Clone(mutation.DeletedNodes),
1150
}
1151
}
1152
1153
assertTargetMap := func(
1154
persistedNodes map[string]*persistencespb.ChasmNode,
1155
mutation NodesMutation,
1156
expected map[string]string,
1157
) {
1158
target, err := s.newTestTree(common.CloneProtoMap(persistedNodes))
1159
s.NoError(err)
1160
component, err := target.Component(NewContext(context.Background(), target), ComponentRef{})
1161
s.NoError(err)
1162
s.Len(component.(*TestComponent).SubComponents, 2, "target parent must be hydrated before replication")
1163
1164
s.NoError(target.ApplyMutation(mutation))
1165
1166
component, err = target.Component(NewContext(context.Background(), target), ComponentRef{})
1167
s.NoError(err)
1168
rootComponent := component.(*TestComponent)
1169
s.Len(rootComponent.SubComponents, len(expected))
1170
for key, runID := range expected {
1171
field, ok := rootComponent.SubComponents[key]
1172
s.True(ok, "expected map key %q", key)
1173
subComponent := field.Get(NewContext(context.Background(), target))
1174
s.Equal(runID, subComponent.SubComponent1Data.GetRunId())
1175
}
1176
}
1177
1178
initialNodes := buildSnapshot(newRootComponent(map[string]string{
1179
"one": "run-one",
1180
"two": "run-two",
1181
}))
1182
1183
s.Run("CreateMapItem", func() {
1184
mutation := mutationFromSource(initialNodes, func(_ Context, component *TestComponent) {
1185
component.SubComponents["three"] = NewComponentField(nil, &TestSubComponent1{
1186
SubComponent1Data: &protoMessageType{RunId: "run-three"},
1187
})
1188
})
1189
1190
assertTargetMap(initialNodes, mutation, map[string]string{
1191
"one": "run-one",
1192
"two": "run-two",
1193
"three": "run-three",
1194
})
1195
})
1196
1197
s.Run("UpdateMapItem", func() {
1198
mutation := mutationFromSource(initialNodes, func(ctx Context, component *TestComponent) {
1199
component.SubComponents["one"].Get(ctx).SubComponent1Data = &protoMessageType{RunId: "run-one-updated"}
1200
})
1201
1202
assertTargetMap(initialNodes, mutation, map[string]string{
1203
"one": "run-one-updated",
1204
"two": "run-two",
1205
})
1206
})
1207
1208
s.Run("DeleteMapItem", func() {
1209
mutation := mutationFromSource(initialNodes, func(_ Context, component *TestComponent) {
1210
delete(component.SubComponents, "one")
1211
})
1212
1213
assertTargetMap(initialNodes, mutation, map[string]string{
1214
"two": "run-two",
1215
})
1216
})
1217
}
1218
1219
func (s *nodeSuite) TestApplyMutation_DeleteUpdateSamePath() {
1220
persistenceNodes := map[string]*persistencespb.ChasmNode{
1221
"": {
1222
Metadata: &persistencespb.ChasmNodeMetadata{
1223
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1224
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1225
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1226
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1227
TypeId: testComponentTypeID,
1228
},
1229
},
1230
},
1231
},
1232
"SubComponent1": {
1233
Metadata: &persistencespb.ChasmNodeMetadata{
1234
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1235
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1236
},
1237
},
1238
}
1239
root, err := s.newTestTree(persistenceNodes)
1240
s.NoError(err)
1241
1242
// First apply a mutation to delete "SubComponent1" node.
1243
err = root.ApplyMutation(NodesMutation{
1244
DeletedNodes: map[string]struct{}{
1245
"SubComponent1": {},
1246
},
1247
})
1248
s.NoError(err)
1249
s.Empty(root.mutation.UpdatedNodes)
1250
s.Len(root.mutation.DeletedNodes, 1)
1251
1252
// Then apply another mutation to update "SubComponent1" node.
1253
// This simulates the applyMutation logic in mutable state where the logic
1254
// first applies a deletion only mutation for recorded chasm node tombstones,
1255
// and then applies an update only mutation for updated nodes.
1256
1257
mutation := NodesMutation{
1258
UpdatedNodes: map[string]*persistencespb.ChasmNode{
1259
"SubComponent1": {
1260
Metadata: &persistencespb.ChasmNodeMetadata{
1261
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 20},
1262
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 20},
1263
},
1264
},
1265
},
1266
}
1267
err = root.ApplyMutation(mutation)
1268
s.NoError(err)
1269
s.Len(root.mutation.UpdatedNodes, 1)
1270
s.Empty(root.mutation.DeletedNodes, 1)
1271
1272
}
1273
1274
func (s *nodeSuite) TestApplySnapshot() {
1275
persistenceNodes := map[string]*persistencespb.ChasmNode{
1276
"": {
1277
Metadata: &persistencespb.ChasmNodeMetadata{
1278
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1279
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1280
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1281
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1282
TypeId: testComponentTypeID,
1283
},
1284
},
1285
},
1286
},
1287
"SubComponent1": {
1288
Metadata: &persistencespb.ChasmNodeMetadata{
1289
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1290
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1291
},
1292
},
1293
"SubComponent1/SubComponent11": {
1294
Metadata: &persistencespb.ChasmNodeMetadata{
1295
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1296
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1297
},
1298
},
1299
"SubComponent1/SubComponent11/SubComponent11Data": {
1300
Metadata: &persistencespb.ChasmNodeMetadata{
1301
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 4},
1302
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 4},
1303
},
1304
},
1305
}
1306
root, err := s.newTestTree(persistenceNodes)
1307
s.NoError(err)
1308
1309
// Set a decoded value that should be reset after applying the snapshot.
1310
root.children["SubComponent1"].value = "decoded-value"
1311
1312
// Prepare an incoming snapshot representing the target state:
1313
// - The "SubComponent1" node is updated (LastUpdateTransition becomes 20),
1314
// - the "SubComponent1/SubComponent11" node is removed,
1315
// - a new node "SubComponent2" is added.
1316
1317
now := timestamppb.Now()
1318
updatedRootData, err := encodeChasmBlob(&protoMessageType{StartTime: now})
1319
s.NoError(err)
1320
incomingSnapshot := NodesSnapshot{
1321
Nodes: map[string]*persistencespb.ChasmNode{
1322
"": {
1323
Metadata: &persistencespb.ChasmNodeMetadata{
1324
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1325
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 10},
1326
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1327
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1328
TypeId: testComponentTypeID,
1329
},
1330
},
1331
},
1332
Data: updatedRootData,
1333
},
1334
"SubComponent1": {
1335
Metadata: &persistencespb.ChasmNodeMetadata{
1336
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1337
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 20},
1338
},
1339
},
1340
"SubComponent2": {
1341
Metadata: &persistencespb.ChasmNodeMetadata{
1342
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 100},
1343
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 100},
1344
},
1345
},
1346
},
1347
}
1348
err = root.ApplySnapshot(incomingSnapshot)
1349
s.NoError(err)
1350
1351
s.Equal(incomingSnapshot, root.Snapshot(nil))
1352
s.Nil(root.children["SubComponent1"].value) // value should be reset after snapshot
1353
1354
// Validate that nodeBase.mutation reflects the applied snapshot.
1355
expectedMutation := NodesMutation{
1356
UpdatedNodes: map[string]*persistencespb.ChasmNode{
1357
"": incomingSnapshot.Nodes[""],
1358
"SubComponent1": incomingSnapshot.Nodes["SubComponent1"],
1359
"SubComponent2": incomingSnapshot.Nodes["SubComponent2"],
1360
},
1361
DeletedNodes: map[string]struct{}{
1362
"SubComponent1/SubComponent11": {},
1363
"SubComponent1/SubComponent11/SubComponent11Data": {},
1364
},
1365
}
1366
s.Equal(expectedMutation, root.mutation)
1367
1368
// Validate visibility search attributes and memo are updated as well.
1369
s.Len(root.currentSA, 3)
1370
s.Contains(root.currentSA, "TemporalDatetime01")
1371
s.True(root.currentSA["TemporalDatetime01"].(VisibilityValueTime).Equal(VisibilityValueTime(now.AsTime())))
1372
}
1373
1374
func (s *nodeSuite) TestApplySnapshot_EmptySnapshot() {
1375
persistenceNodes := map[string]*persistencespb.ChasmNode{
1376
"": {
1377
Metadata: &persistencespb.ChasmNodeMetadata{
1378
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1379
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1380
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1381
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1382
TypeId: testComponentTypeID,
1383
},
1384
},
1385
},
1386
},
1387
"SubComponent1": {
1388
Metadata: &persistencespb.ChasmNodeMetadata{
1389
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1390
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1391
},
1392
},
1393
}
1394
root, err := s.newTestTree(persistenceNodes)
1395
s.NoError(err)
1396
1397
// Apply an empty snapshot to simulate the case where
1398
// chasm is disabled in source cluster or chasm tree is empty in
1399
// source cluster.
1400
err = root.ApplySnapshot(NodesSnapshot{})
1401
s.NoError(err)
1402
1403
// Validate that nodeBase.mutation reflects the applied snapshot.
1404
expectedMutation := NodesMutation{
1405
UpdatedNodes: map[string]*persistencespb.ChasmNode{},
1406
DeletedNodes: map[string]struct{}{
1407
"SubComponent1": {}, // NOTE: root component can't be deleted.
1408
},
1409
}
1410
s.Equal(expectedMutation, root.mutation)
1411
}
1412
1413
func (s *nodeSuite) TestApplyMutation_OutOfOrder() {
1414
persistenceNodes := map[string]*persistencespb.ChasmNode{
1415
"": {
1416
Metadata: &persistencespb.ChasmNodeMetadata{
1417
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1418
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1419
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1420
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1421
TypeId: testComponentTypeID,
1422
},
1423
},
1424
},
1425
},
1426
}
1427
1428
root, err := s.newTestTree(persistenceNodes)
1429
s.NoError(err)
1430
1431
// Test the case where child node is applied before parent node.
1432
err = root.ApplyMutation(NodesMutation{
1433
UpdatedNodes: map[string]*persistencespb.ChasmNode{
1434
"SubComponent1/SubComponent11": {
1435
Metadata: &persistencespb.ChasmNodeMetadata{
1436
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1437
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 20},
1438
},
1439
},
1440
},
1441
})
1442
s.NoError(err)
1443
1444
err = root.ApplyMutation(NodesMutation{
1445
UpdatedNodes: map[string]*persistencespb.ChasmNode{
1446
"": {
1447
Metadata: &persistencespb.ChasmNodeMetadata{
1448
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1449
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1450
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1451
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1452
TypeId: testComponentTypeID,
1453
},
1454
},
1455
},
1456
},
1457
"SubComponent1": {
1458
Metadata: &persistencespb.ChasmNodeMetadata{
1459
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1460
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1461
},
1462
},
1463
},
1464
})
1465
s.NoError(err)
1466
1467
snapshot := root.Snapshot(nil)
1468
s.Len(snapshot.Nodes, 3)
1469
s.Len(root.mutation.UpdatedNodes, 3)
1470
}
1471
1472
func (s *nodeSuite) partitionedSnapshotTestNodes() map[string]*persistencespb.ChasmNode {
1473
return map[string]*persistencespb.ChasmNode{
1474
"": {
1475
Metadata: &persistencespb.ChasmNodeMetadata{
1476
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1477
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1478
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1479
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1480
TypeId: testComponentTypeID,
1481
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{
1482
{
1483
TypeId: testSideEffectTaskTypeID,
1484
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1485
VersionedTransitionOffset: 1,
1486
PhysicalTaskStatus: physicalTaskStatusCreated,
1487
},
1488
{
1489
TypeId: testSideEffectTaskTypeID,
1490
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1491
VersionedTransitionOffset: 2,
1492
PhysicalTaskStatus: physicalTaskStatusNone,
1493
},
1494
},
1495
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
1496
{
1497
TypeId: testPureTaskTypeID,
1498
ScheduledTime: timestamppb.New(s.timeSource.Now().Add(time.Minute)),
1499
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1500
VersionedTransitionOffset: 3,
1501
PhysicalTaskStatus: physicalTaskStatusCreated,
1502
},
1503
},
1504
},
1505
},
1506
},
1507
},
1508
// A component node with no tasks carries no cluster-local metadata and must be skipped.
1509
"SubComponent1": {
1510
Metadata: &persistencespb.ChasmNodeMetadata{
1511
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1512
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1513
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1514
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1515
TypeId: testSubComponent1TypeID,
1516
},
1517
},
1518
},
1519
},
1520
}
1521
}
1522
1523
func (s *nodeSuite) TestPartitionedSnapshot() {
1524
root, err := s.newTestTree(s.partitionedSnapshotTestNodes())
1525
s.NoError(err)
1526
1527
clean, localState := root.PartitionedSnapshot(nil)
1528
1529
// Cluster-local state captures only the component node with tasks, in order.
1530
s.Len(localState.GetNodes(), 1)
1531
rootState := localState.GetNodes()[""]
1532
s.NotNil(rootState)
1533
s.Equal([]int32{physicalTaskStatusCreated, physicalTaskStatusNone}, rootState.GetSideEffectTaskStatuses())
1534
s.Equal([]int32{physicalTaskStatusCreated}, rootState.GetPureTaskStatuses())
1535
1536
// The clean snapshot keeps every node key but zeroes the cluster-local fields.
1537
s.Len(clean.Nodes, 2)
1538
cleanAttr := clean.Nodes[""].GetMetadata().GetComponentAttributes()
1539
for _, t := range cleanAttr.GetSideEffectTasks() {
1540
s.Equal(physicalTaskStatusNone, t.GetPhysicalTaskStatus())
1541
}
1542
for _, t := range cleanAttr.GetPureTasks() {
1543
s.Equal(physicalTaskStatusNone, t.GetPhysicalTaskStatus())
1544
}
1545
1546
// The live tree must be untouched: Snapshot returns the original statuses.
1547
liveAttr := root.Snapshot(nil).Nodes[""].GetMetadata().GetComponentAttributes()
1548
s.Equal(physicalTaskStatusCreated, liveAttr.GetSideEffectTasks()[0].GetPhysicalTaskStatus())
1549
s.Equal(physicalTaskStatusCreated, liveAttr.GetPureTasks()[0].GetPhysicalTaskStatus())
1550
}
1551
1552
func (s *nodeSuite) TestPartitionedSnapshot_MergeRoundTrip() {
1553
root, err := s.newTestTree(s.partitionedSnapshotTestNodes())
1554
s.NoError(err)
1555
1556
clean, localState := root.PartitionedSnapshot(nil)
1557
1558
// Merging the extracted state back into the clean snapshot restores the statuses with no mismatch.
1559
s.Zero(clean.MergeClusterLocalState(localState))
1560
mergedAttr := clean.Nodes[""].GetMetadata().GetComponentAttributes()
1561
s.Equal(physicalTaskStatusCreated, mergedAttr.GetSideEffectTasks()[0].GetPhysicalTaskStatus())
1562
s.Equal(physicalTaskStatusNone, mergedAttr.GetSideEffectTasks()[1].GetPhysicalTaskStatus())
1563
s.Equal(physicalTaskStatusCreated, mergedAttr.GetPureTasks()[0].GetPhysicalTaskStatus())
1564
}
1565
1566
func (s *nodeSuite) TestMergeClusterLocalState_ReportsLengthMismatch() {
1567
root, err := s.newTestTree(s.partitionedSnapshotTestNodes())
1568
s.NoError(err)
1569
1570
clean, localState := root.PartitionedSnapshot(nil)
1571
1572
// Truncate the root node's side-effect statuses so there are fewer statuses than tasks. The
1573
// uncovered tasks stay zeroed — the benign, self-healing direction.
1574
localState.Nodes[""].SideEffectTaskStatuses = localState.Nodes[""].SideEffectTaskStatuses[:1]
1575
s.Equal(ClusterLocalStateMergeResult{NodesWithUncoveredTasks: 1}, clean.MergeClusterLocalState(localState))
1576
}
1577
1578
func (s *nodeSuite) TestMergeClusterLocalState_ReportsExtraStatuses() {
1579
root, err := s.newTestTree(s.partitionedSnapshotTestNodes())
1580
s.NoError(err)
1581
1582
clean, localState := root.PartitionedSnapshot(nil)
1583
1584
// Append a surplus side-effect status so there are more statuses than tasks. The extra status
1585
// has no task to apply to and is dropped — the suspicious (possible-divergence) direction.
1586
localState.Nodes[""].SideEffectTaskStatuses = append(localState.Nodes[""].SideEffectTaskStatuses, physicalTaskStatusCreated)
1587
s.Equal(ClusterLocalStateMergeResult{NodesWithExtraStatuses: 1}, clean.MergeClusterLocalState(localState))
1588
}
1589
1590
func (s *nodeSuite) TestMergeClusterLocalState_SkipsMissingNodes() {
1591
root, err := s.newTestTree(s.partitionedSnapshotTestNodes())
1592
s.NoError(err)
1593
1594
clean, localState := root.PartitionedSnapshot(nil)
1595
1596
// Simulate the component node being deleted from the snapshot before merge.
1597
delete(clean.Nodes, "")
1598
s.NotPanics(func() {
1599
clean.MergeClusterLocalState(localState)
1600
})
1601
s.Len(clean.Nodes, 1)
1602
}
1603
1604
// A tree whose nodes carry no tasks has no cluster-local state: PartitionedSnapshot returns a
1605
// state with an empty Nodes map, and merging that empty state back is a no-op.
1606
func (s *nodeSuite) TestPartitionedSnapshot_NoClusterLocalState() {
1607
root, err := s.newTestTree(map[string]*persistencespb.ChasmNode{
1608
"": {
1609
Metadata: &persistencespb.ChasmNodeMetadata{
1610
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1611
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1612
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1613
ComponentAttributes: &persistencespb.ChasmComponentAttributes{TypeId: testComponentTypeID},
1614
},
1615
},
1616
},
1617
})
1618
s.NoError(err)
1619
1620
clean, localState := root.PartitionedSnapshot(nil)
1621
s.NotNil(localState)
1622
s.Empty(localState.Nodes)
1623
s.Len(clean.Nodes, 1)
1624
s.Zero(clean.MergeClusterLocalState(localState))
1625
}
1626
1627
// TestPartitionedSnapshot_ClusterLocalFieldGuard is a tripwire against silent drift in the set of
1628
// cluster-local CHASM fields. Cluster-local state — currently only
1629
// ChasmComponentAttributes.Task.physical_task_status — must be extracted by PartitionedSnapshot
1630
// before upload/replication and restored by MergeClusterLocalState on read; otherwise it leaks
1631
// across clusters. The partition logic only inspects component_attributes' side_effect_tasks and
1632
// pure_tasks, so a new field on any message below — or a new attribute type in the
1633
// ChasmNodeMetadata oneof — could introduce cluster-local state the logic silently misses.
1634
//
1635
// When this fails: decide whether the added/removed field is cluster-local. If it is, handle it in
1636
// PartitionedSnapshot + MergeClusterLocalState and extend ChasmLocalState. Either way, update
1637
// the expected field set below once the partition logic is confirmed correct.
1638
func TestPartitionedSnapshot_ClusterLocalFieldGuard(t *testing.T) {
1639
t.Parallel()
1640
1641
cases := []struct {
1642
msg proto.Message
1643
want []string
1644
}{
1645
// The node wrapper itself: a cluster-local field added directly to the node (rather than its
1646
// metadata) must still trip the guard.
1647
{&persistencespb.ChasmNode{}, []string{"metadata", "data"}},
1648
{
1649
&persistencespb.ChasmNodeMetadata{},
1650
[]string{
1651
"initial_versioned_transition",
1652
"last_update_versioned_transition",
1653
"component_attributes",
1654
"data_attributes",
1655
"collection_attributes",
1656
"pointer_attributes",
1657
},
1658
},
1659
{
1660
&persistencespb.ChasmComponentAttributes{},
1661
[]string{"type_id", "side_effect_tasks", "pure_tasks", "detached", "requests", "user_metadata"},
1662
},
1663
{
1664
// The only message carrying a cluster-local field today (physical_task_status).
1665
&persistencespb.ChasmComponentAttributes_Task{},
1666
[]string{
1667
"type_id", "destination", "scheduled_time", "data",
1668
"versioned_transition", "versioned_transition_offset", "physical_task_status",
1669
},
1670
},
1671
// Reachable via ChasmComponentAttributes.requests.
1672
{&persistencespb.ChasmComponentAttributes_RequestMetadata{}, []string{"links"}},
1673
// Attribute types other than component carry no cluster-local state today; pinning their
1674
// fields ensures a future cluster-local field added to one of them trips this guard too.
1675
{&persistencespb.ChasmDataAttributes{}, nil},
1676
{&persistencespb.ChasmCollectionAttributes{}, nil},
1677
{&persistencespb.ChasmPointerAttributes{}, []string{"node_path"}},
1678
}
1679
1680
for _, tc := range cases {
1681
desc := tc.msg.ProtoReflect().Descriptor()
1682
fields := desc.Fields()
1683
got := make([]string, 0, fields.Len())
1684
for i := 0; i < fields.Len(); i++ {
1685
got = append(got, string(fields.Get(i).Name()))
1686
}
1687
require.ElementsMatchf(t, tc.want, got,
1688
"%s field set changed; cluster-local partition logic in PartitionedSnapshot/"+
1689
"MergeClusterLocalState may be stale (see this test's doc comment)", desc.FullName())
1690
}
1691
}
1692
1693
func (s *nodeSuite) TestRefreshTasks() {
1694
now := s.timeSource.Now()
1695
pureTaskScheduledTime := now.Add(time.Second).UTC()
1696
persistenceNodes := map[string]*persistencespb.ChasmNode{
1697
"": {
1698
Metadata: &persistencespb.ChasmNodeMetadata{
1699
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1700
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1701
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1702
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1703
TypeId: testComponentTypeID,
1704
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
1705
{
1706
TypeId: testPureTaskTypeID,
1707
ScheduledTime: timestamppb.New(now.Add(time.Minute)),
1708
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1709
VersionedTransitionOffset: 1,
1710
PhysicalTaskStatus: physicalTaskStatusNone,
1711
},
1712
},
1713
},
1714
},
1715
},
1716
},
1717
"SubComponent1": {
1718
Metadata: &persistencespb.ChasmNodeMetadata{
1719
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1720
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1721
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1722
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1723
TypeId: testSubComponent1TypeID,
1724
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
1725
{
1726
TypeId: testPureTaskTypeID,
1727
ScheduledTime: timestamppb.New(pureTaskScheduledTime),
1728
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1729
VersionedTransitionOffset: 2,
1730
PhysicalTaskStatus: physicalTaskStatusCreated,
1731
},
1732
},
1733
},
1734
},
1735
},
1736
},
1737
"SubComponent2": {
1738
Metadata: &persistencespb.ChasmNodeMetadata{
1739
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1740
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1741
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1742
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1743
TypeId: testSubComponent2TypeID,
1744
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{
1745
{
1746
TypeId: testSideEffectTaskTypeID,
1747
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1748
VersionedTransitionOffset: 3,
1749
PhysicalTaskStatus: physicalTaskStatusCreated,
1750
},
1751
},
1752
},
1753
},
1754
},
1755
},
1756
}
1757
1758
root, err := s.newTestTree(persistenceNodes)
1759
s.NoError(err)
1760
1761
err = root.RefreshTasks()
1762
s.NoError(err)
1763
1764
s.True(root.IsDirty())
1765
s.False(root.IsStateDirty())
1766
1767
mutation, err := root.CloseTransaction()
1768
s.NoError(err)
1769
s.Len(mutation.UpdatedNodes, 2) // TaskStatus for the root node is not reset, so no need to persist it.
1770
s.Equal(2, s.nodeBackend.NumTasksAdded())
1771
s.Equal(pureTaskScheduledTime, s.nodeBackend.LastDeletePureTaskCall())
1772
}
1773
1774
func (s *nodeSuite) TestCarryOverTaskStatus() {
1775
now := s.timeSource.Now()
1776
persistenceNodes := map[string]*persistencespb.ChasmNode{
1777
"": {
1778
Metadata: &persistencespb.ChasmNodeMetadata{
1779
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1780
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1781
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1782
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1783
TypeId: testComponentTypeID,
1784
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{
1785
{
1786
TypeId: testSideEffectTaskTypeID,
1787
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1788
VersionedTransitionOffset: 1,
1789
PhysicalTaskStatus: physicalTaskStatusCreated,
1790
},
1791
{
1792
TypeId: testSideEffectTaskTypeID,
1793
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1794
VersionedTransitionOffset: 2,
1795
PhysicalTaskStatus: physicalTaskStatusCreated,
1796
},
1797
{
1798
TypeId: testSideEffectTaskTypeID,
1799
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1800
VersionedTransitionOffset: 1,
1801
PhysicalTaskStatus: physicalTaskStatusCreated,
1802
},
1803
},
1804
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
1805
{
1806
TypeId: testPureTaskTypeID,
1807
ScheduledTime: timestamppb.New(now.Add(time.Minute)),
1808
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1809
VersionedTransitionOffset: 2,
1810
PhysicalTaskStatus: physicalTaskStatusCreated,
1811
},
1812
{
1813
TypeId: testPureTaskTypeID,
1814
ScheduledTime: timestamppb.New(now.Add(2 * time.Minute)),
1815
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1816
VersionedTransitionOffset: 3,
1817
PhysicalTaskStatus: physicalTaskStatusCreated,
1818
},
1819
{
1820
TypeId: testPureTaskTypeID,
1821
ScheduledTime: timestamppb.New(now.Add(3 * time.Minute)),
1822
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1823
VersionedTransitionOffset: 3,
1824
PhysicalTaskStatus: physicalTaskStatusNone,
1825
},
1826
},
1827
},
1828
},
1829
},
1830
},
1831
"data": {
1832
Metadata: &persistencespb.ChasmNodeMetadata{
1833
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1834
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1835
Attributes: &persistencespb.ChasmNodeMetadata_DataAttributes{
1836
DataAttributes: &persistencespb.ChasmDataAttributes{},
1837
},
1838
},
1839
},
1840
}
1841
root, err := s.newTestTree(persistenceNodes)
1842
s.NoError(err)
1843
1844
mutations := NodesMutation{
1845
UpdatedNodes: map[string]*persistencespb.ChasmNode{
1846
"": {
1847
Metadata: &persistencespb.ChasmNodeMetadata{
1848
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1849
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1850
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1851
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1852
TypeId: testComponentTypeID,
1853
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{
1854
{
1855
TypeId: testSideEffectTaskTypeID,
1856
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1857
VersionedTransitionOffset: 2,
1858
PhysicalTaskStatus: physicalTaskStatusCreated,
1859
},
1860
{
1861
TypeId: testSideEffectTaskTypeID,
1862
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1863
VersionedTransitionOffset: 1,
1864
PhysicalTaskStatus: physicalTaskStatusCreated,
1865
},
1866
},
1867
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
1868
{
1869
TypeId: testPureTaskTypeID,
1870
ScheduledTime: timestamppb.New(now.Add(time.Second)),
1871
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1872
VersionedTransitionOffset: 2,
1873
PhysicalTaskStatus: physicalTaskStatusCreated,
1874
},
1875
{
1876
TypeId: testPureTaskTypeID,
1877
ScheduledTime: timestamppb.New(now.Add(2 * time.Minute)),
1878
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1879
VersionedTransitionOffset: 3,
1880
PhysicalTaskStatus: physicalTaskStatusCreated,
1881
},
1882
{
1883
TypeId: testPureTaskTypeID,
1884
ScheduledTime: timestamppb.New(now.Add(3 * time.Minute)),
1885
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1886
VersionedTransitionOffset: 3,
1887
PhysicalTaskStatus: physicalTaskStatusNone,
1888
},
1889
},
1890
},
1891
},
1892
},
1893
},
1894
"data": {
1895
Metadata: &persistencespb.ChasmNodeMetadata{
1896
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1897
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1898
Attributes: &persistencespb.ChasmNodeMetadata_DataAttributes{
1899
DataAttributes: &persistencespb.ChasmDataAttributes{},
1900
},
1901
},
1902
},
1903
},
1904
}
1905
1906
expectedNodes := map[string]*persistencespb.ChasmNode{
1907
"": {
1908
Metadata: &persistencespb.ChasmNodeMetadata{
1909
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1910
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1911
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
1912
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
1913
TypeId: testComponentTypeID,
1914
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{
1915
{
1916
TypeId: testSideEffectTaskTypeID,
1917
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1918
VersionedTransitionOffset: 2,
1919
PhysicalTaskStatus: physicalTaskStatusCreated,
1920
},
1921
{
1922
TypeId: testSideEffectTaskTypeID,
1923
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1924
VersionedTransitionOffset: 1,
1925
PhysicalTaskStatus: physicalTaskStatusNone,
1926
},
1927
},
1928
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
1929
{
1930
TypeId: testPureTaskTypeID,
1931
ScheduledTime: timestamppb.New(now.Add(time.Second)),
1932
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1933
VersionedTransitionOffset: 2,
1934
PhysicalTaskStatus: physicalTaskStatusNone,
1935
},
1936
{
1937
TypeId: testPureTaskTypeID,
1938
ScheduledTime: timestamppb.New(now.Add(2 * time.Minute)),
1939
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
1940
VersionedTransitionOffset: 3,
1941
PhysicalTaskStatus: physicalTaskStatusCreated,
1942
},
1943
{
1944
TypeId: testPureTaskTypeID,
1945
ScheduledTime: timestamppb.New(now.Add(3 * time.Minute)),
1946
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1947
VersionedTransitionOffset: 3,
1948
PhysicalTaskStatus: physicalTaskStatusNone,
1949
},
1950
},
1951
},
1952
},
1953
},
1954
},
1955
"data": {
1956
Metadata: &persistencespb.ChasmNodeMetadata{
1957
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
1958
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 3},
1959
Attributes: &persistencespb.ChasmNodeMetadata_DataAttributes{
1960
DataAttributes: &persistencespb.ChasmDataAttributes{},
1961
},
1962
},
1963
},
1964
}
1965
1966
err = root.ApplyMutation(mutations)
1967
s.NoError(err)
1968
1969
s.Equal(expectedNodes, root.Snapshot(nil).Nodes)
1970
}
1971
1972
func (s *nodeSuite) TestValidateAccess() {
1973
nodePath := []string{"SubComponent1", "SubComponent11"}
1974
1975
// Because access checks are performed on ancestor nodes and not the target node,
1976
// test case properties are applied to the root node.
1977
testCases := []struct {
1978
name string
1979
valid bool
1980
intent OperationIntent
1981
componentStatus enumspb.WorkflowExecutionStatus // TestComponent borrows the WorkflowExecutionStatus struct
1982
executionStatus enumspb.WorkflowExecutionStatus
1983
executionState enumsspb.WorkflowExecutionState
1984
terminated bool
1985
1986
setup func(*Node, Context) error
1987
}{
1988
{
1989
name: "access check applies only to ancestors (terminated)",
1990
valid: true,
1991
intent: OperationIntentProgress,
1992
componentStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
1993
executionStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
1994
executionState: enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
1995
terminated: false,
1996
setup: func(target *Node, ctx Context) error {
1997
// Set the terminated flag on the target node instead of an ancestor
1998
target.terminated = true
1999
return nil
2000
},
2001
},
2002
{
2003
name: "access check applies only to ancestors (closed)",
2004
valid: true,
2005
intent: OperationIntentProgress,
2006
componentStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
2007
executionStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
2008
executionState: enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
2009
terminated: false,
2010
setup: func(target *Node, ctx Context) error {
2011
if err := target.prepareComponentValue(ctx); err != nil {
2012
return err
2013
}
2014
targetComponent, _ := target.value.(*TestSubComponent11)
2015
targetComponent.SubComponent11Data.Status = enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
2016
return nil
2017
},
2018
},
2019
{
2020
name: "read-only always succeeds",
2021
intent: OperationIntentObserve,
2022
componentStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
2023
executionStatus: enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED,
2024
executionState: enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
2025
terminated: true,
2026
valid: true,
2027
},
2028
{
2029
name: "valid write access",
2030
intent: OperationIntentProgress,
2031
componentStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
2032
executionStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
2033
executionState: enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
2034
terminated: false,
2035
valid: true,
2036
},
2037
{
2038
name: "invalid write access (parent closed)",
2039
intent: OperationIntentProgress,
2040
componentStatus: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
2041
executionStatus: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
2042
executionState: enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
2043
terminated: false,
2044
valid: false,
2045
},
2046
{
2047
name: "invalid write access (component terminated)",
2048
intent: OperationIntentProgress,
2049
componentStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
2050
executionStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
2051
executionState: enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
2052
terminated: true, // terminated in current transaction
2053
valid: false,
2054
},
2055
{
2056
name: "invalid write access (component terminated and reload)",
2057
intent: OperationIntentProgress,
2058
componentStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
2059
executionStatus: enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED,
2060
executionState: enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
2061
terminated: false, // terminated in previous transaction and mutable state reloaded
2062
valid: false,
2063
},
2064
{
2065
name: "detached node skips parent validation",
2066
valid: true,
2067
intent: OperationIntentProgress,
2068
componentStatus: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
2069
executionStatus: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, // root is closed
2070
executionState: enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
2071
terminated: false,
2072
setup: func(target *Node, _ Context) error {
2073
// Set the parent node (SubComponent1) as detached.
2074
// When validateParentAccess is called on a detached node, it skips
2075
// ancestor validation entirely.
2076
target.parent.serializedNode.GetMetadata().GetComponentAttributes().Detached = true
2077
return nil
2078
},
2079
},
2080
}
2081
2082
for _, tc := range testCases {
2083
s.Run(tc.name, func() {
2084
root, err := s.newTestTree(testComponentSerializedNodes())
2085
s.NoError(err)
2086
2087
ctx := NewContext(
2088
newContextWithOperationIntent(context.Background(), tc.intent),
2089
root,
2090
)
2091
2092
// Set fields on root node
2093
err = root.prepareComponentValue(ctx)
2094
s.NoError(err)
2095
root.terminated = tc.terminated
2096
component, ok := root.value.(*TestComponent)
2097
if ok {
2098
component.ComponentData.Status = tc.componentStatus
2099
}
2100
2101
// Find target node
2102
node, ok := root.findNode(nodePath)
2103
s.True(ok)
2104
err = node.prepareComponentValue(ctx)
2105
s.NoError(err)
2106
2107
if tc.setup != nil {
2108
s.NoError(tc.setup(node, ctx))
2109
}
2110
2111
s.nodeBackend.HandleGetExecutionState = func() *persistencespb.WorkflowExecutionState {
2112
return &persistencespb.WorkflowExecutionState{
2113
State: tc.executionState,
2114
Status: tc.executionStatus,
2115
}
2116
}
2117
2118
// Validation begins on the target node, checking ancestors only.
2119
err = node.validateAccess(ctx, false)
2120
if tc.valid {
2121
s.NoError(err)
2122
} else {
2123
s.Error(err)
2124
s.ErrorIs(errAccessCheckFailed, err)
2125
}
2126
})
2127
}
2128
2129
}
2130
2131
func (s *nodeSuite) TestGetComponent_DetachedNodeBypassesParentValidation() {
2132
// Test that a detached node can be accessed even when its parent is closed.
2133
root, err := s.newTestTree(testComponentSerializedNodes())
2134
s.NoError(err)
2135
2136
targetPath := []string{"SubComponent1", "SubComponent11"}
2137
targetNode, ok := root.findNode(targetPath)
2138
s.True(ok)
2139
2140
// Mark the target node as detached.
2141
targetNode.serializedNode.GetMetadata().GetComponentAttributes().Detached = true
2142
2143
// Close the root node (set lifecycle to COMPLETED).
2144
ctx := NewMutableContext(
2145
newContextWithOperationIntent(context.Background(), OperationIntentProgress),
2146
root,
2147
)
2148
err = root.prepareComponentValue(ctx)
2149
s.NoError(err)
2150
rootComponent, ok := root.value.(*TestComponent)
2151
s.True(ok)
2152
rootComponent.ComponentData.Status = enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
2153
2154
// GetComponent on the detached node should succeed despite root being closed.
2155
ref := ComponentRef{
2156
componentPath: targetPath,
2157
}
2158
component, err := root.Component(ctx, ref)
2159
s.NoError(err)
2160
s.NotNil(component)
2161
}
2162
2163
func (s *nodeSuite) TestGetComponent_ClosedTargetSucceeds() {
2164
// Test that a closed target component can still be accessed via Component()
2165
// because we only check ancestor lifecycle, not the target's lifecycle.
2166
root, err := s.newTestTree(testComponentSerializedNodes())
2167
s.NoError(err)
2168
2169
targetPath := []string{"SubComponent1", "SubComponent11"}
2170
targetNode, ok := root.findNode(targetPath)
2171
s.True(ok)
2172
2173
ctx := NewMutableContext(
2174
newContextWithOperationIntent(context.Background(), OperationIntentProgress),
2175
root,
2176
)
2177
2178
// Close the target node's lifecycle (set to COMPLETED).
2179
err = targetNode.prepareComponentValue(ctx)
2180
s.NoError(err)
2181
targetComponent, ok := targetNode.value.(*TestSubComponent11)
2182
s.True(ok)
2183
targetComponent.SubComponent11Data.Status = enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
2184
s.True(targetComponent.LifecycleState(ctx).IsClosed())
2185
2186
// GetComponent on the closed target should succeed because we only check ancestors.
2187
ref := ComponentRef{
2188
componentPath: targetPath,
2189
}
2190
component, err := root.Component(ctx, ref)
2191
s.NoError(err)
2192
s.NotNil(component)
2193
}
2194
2195
func (s *nodeSuite) TestGetComponent() {
2196
errValidation := errors.New("some random validation error")
2197
2198
expectedTestComponent := &TestComponent{}
2199
setTestComponentFields(expectedTestComponent, s.nodeBackend)
2200
assertTestComponent := func(component Component) {
2201
testComponent, ok := component.(*TestComponent)
2202
s.True(ok)
2203
protoassert.ProtoEqual(s.T(), expectedTestComponent.ComponentData, testComponent.ComponentData)
2204
2205
// TODO: Can we assert other fields?
2206
// Right now the chasm Field generated by setTestComponentFields() doesn't have a backing node.
2207
}
2208
2209
testCases := []struct {
2210
name string
2211
chasmContextFn func(root *Node) Context
2212
ref ComponentRef
2213
expectedErr error
2214
nodeDirty bool
2215
assertComponent func(Component)
2216
}{
2217
{
2218
name: "path not found",
2219
chasmContextFn: func(root *Node) Context {
2220
return NewContext(context.Background(), root)
2221
},
2222
ref: ComponentRef{
2223
componentPath: []string{"unknownComponent"},
2224
},
2225
expectedErr: errComponentNotFound,
2226
},
2227
{
2228
name: "initialVT mismatch",
2229
chasmContextFn: func(root *Node) Context {
2230
return NewMutableContext(context.Background(), root)
2231
},
2232
ref: ComponentRef{
2233
componentPath: []string{"SubComponent1", "SubComponent11"},
2234
// should be (1, 1) but we set it to (2, 2)
2235
componentInitialVT: &persistencespb.VersionedTransition{
2236
NamespaceFailoverVersion: 2,
2237
TransitionCount: 2,
2238
},
2239
},
2240
expectedErr: errComponentNotFound,
2241
},
2242
{
2243
name: "validation failure",
2244
chasmContextFn: func(root *Node) Context {
2245
return NewMutableContext(context.Background(), root)
2246
},
2247
ref: ComponentRef{
2248
componentPath: []string{"SubComponent1"},
2249
componentInitialVT: &persistencespb.VersionedTransition{
2250
NamespaceFailoverVersion: 1,
2251
TransitionCount: 1,
2252
},
2253
validationFn: func(_ NodeBackend, _ Context, _ Component, _ *Registry) error {
2254
return errValidation
2255
},
2256
},
2257
expectedErr: errValidation,
2258
},
2259
{
2260
name: "success readonly access",
2261
chasmContextFn: func(root *Node) Context {
2262
return NewContext(context.Background(), root)
2263
},
2264
ref: ComponentRef{
2265
componentPath: []string{}, // root
2266
componentInitialVT: &persistencespb.VersionedTransition{
2267
NamespaceFailoverVersion: 1,
2268
TransitionCount: 1,
2269
},
2270
validationFn: func(_ NodeBackend, _ Context, _ Component, _ *Registry) error {
2271
return nil
2272
},
2273
},
2274
expectedErr: nil,
2275
assertComponent: assertTestComponent,
2276
},
2277
{
2278
name: "success mutable access",
2279
chasmContextFn: func(root *Node) Context {
2280
return NewMutableContext(context.Background(), root)
2281
},
2282
ref: ComponentRef{
2283
componentPath: []string{}, // root
2284
},
2285
expectedErr: nil,
2286
nodeDirty: true,
2287
assertComponent: assertTestComponent,
2288
},
2289
}
2290
2291
for _, tc := range testCases {
2292
s.Run(tc.name, func() {
2293
root, err := s.newTestTree(testComponentSerializedNodes())
2294
s.NoError(err)
2295
2296
component, err := root.Component(tc.chasmContextFn(root), tc.ref)
2297
s.Equal(tc.expectedErr, err)
2298
2299
node, ok := root.findNode(tc.ref.componentPath)
2300
if tc.expectedErr == nil {
2301
s.True(ok)
2302
tc.assertComponent(component)
2303
}
2304
2305
if ok {
2306
if tc.nodeDirty {
2307
s.Greater(node.valueState, valueStateSynced)
2308
} else {
2309
s.LessOrEqual(node.valueState, valueStateSynced)
2310
}
2311
}
2312
})
2313
}
2314
}
2315
2316
func (s *nodeSuite) TestRef() {
2317
workflowKey := definition.NewWorkflowKey(
2318
primitives.NewUUID().String(),
2319
primitives.NewUUID().String(),
2320
primitives.NewUUID().String(),
2321
)
2322
executionKey := ExecutionKey{
2323
NamespaceID: workflowKey.NamespaceID,
2324
BusinessID: workflowKey.WorkflowID,
2325
RunID: workflowKey.RunID,
2326
}
2327
currentVT := &persistencespb.VersionedTransition{
2328
NamespaceFailoverVersion: 2,
2329
TransitionCount: 2,
2330
}
2331
s.nodeBackend = &MockNodeBackend{
2332
HandleCurrentVersionedTransition: func() *persistencespb.VersionedTransition {
2333
return currentVT
2334
},
2335
HandleGetWorkflowKey: func() definition.WorkflowKey {
2336
return workflowKey
2337
},
2338
}
2339
2340
root, err := s.newTestTree(testComponentSerializedNodes())
2341
s.NoError(err)
2342
2343
chasmContext := NewContext(context.Background(), root)
2344
rootComponent, err := root.Component(chasmContext, NewComponentRef[*TestComponent](executionKey))
2345
s.NoError(err)
2346
testComponent, ok := rootComponent.(*TestComponent)
2347
s.True(ok)
2348
2349
rc, ok := s.registry.ComponentFor(testComponent)
2350
s.True(ok)
2351
archetypeID := rc.componentID
2352
2353
subComponent1 := testComponent.SubComponent1.Get(chasmContext)
2354
subComponent11 := subComponent1.SubComponent11.Get(chasmContext)
2355
2356
testCases := []struct {
2357
name string
2358
component Component
2359
expectErr bool
2360
expectedPath []string
2361
expectedInitalVT *persistencespb.VersionedTransition
2362
}{
2363
{
2364
name: "root",
2365
component: testComponent,
2366
expectErr: false,
2367
expectedPath: nil, // same as []string{}
2368
expectedInitalVT: &persistencespb.VersionedTransition{
2369
NamespaceFailoverVersion: 1,
2370
TransitionCount: 1,
2371
},
2372
},
2373
{
2374
name: "subComponent1",
2375
component: subComponent1,
2376
expectErr: false,
2377
expectedPath: []string{"SubComponent1"},
2378
expectedInitalVT: &persistencespb.VersionedTransition{
2379
NamespaceFailoverVersion: 1,
2380
TransitionCount: 1,
2381
},
2382
},
2383
{
2384
name: "subComponent11",
2385
component: subComponent11,
2386
expectErr: false,
2387
expectedPath: []string{"SubComponent1", "SubComponent11"},
2388
expectedInitalVT: &persistencespb.VersionedTransition{
2389
NamespaceFailoverVersion: 1,
2390
TransitionCount: 1,
2391
},
2392
},
2393
{
2394
name: "unknown",
2395
component: &TestComponent{}, // a new instance of TestComponent
2396
expectErr: true,
2397
},
2398
}
2399
2400
for _, tc := range testCases {
2401
s.Run(tc.name, func() {
2402
2403
encodedRef, err := root.Ref(tc.component)
2404
if tc.expectErr {
2405
s.Error(err)
2406
return
2407
}
2408
2409
s.NoError(err)
2410
expectedRef := ComponentRef{
2411
ExecutionKey: executionKey,
2412
archetypeID: archetypeID,
2413
componentPath: tc.expectedPath,
2414
2415
// Proto fields are validated separately with ProtoEqual.
2416
// executionLastUpdateVT: currentVT,
2417
// componentInitialVT: tc.expectedInitalVT,
2418
}
2419
2420
actualRef, err := DeserializeComponentRef(encodedRef)
2421
s.NoError(err)
2422
s.ProtoEqual(currentVT, actualRef.executionLastUpdateVT)
2423
s.ProtoEqual(tc.expectedInitalVT, actualRef.componentInitialVT)
2424
2425
actualRef.executionLastUpdateVT = nil
2426
actualRef.componentInitialVT = nil
2427
s.Equal(expectedRef, actualRef)
2428
})
2429
}
2430
}
2431
2432
func (s *nodeSuite) TestSerializeDeserializeTask() {
2433
payload := &commonpb.Payload{
2434
Data: []byte("some-random-data"),
2435
}
2436
expectedBlob, err := encodeChasmBlob(payload)
2437
s.NoError(err)
2438
2439
testCases := []struct {
2440
name string
2441
task any
2442
expectedData []byte
2443
equalFn func(t1, t2 any)
2444
}{
2445
{
2446
name: "ProtoTask",
2447
task: &TestSideEffectTask{
2448
Data: []byte("some-random-data"),
2449
},
2450
expectedData: expectedBlob.GetData(),
2451
equalFn: func(t1, t2 any) {
2452
protorequire.ProtoEqual(s.T(), t1.(*TestSideEffectTask), t2.(*TestSideEffectTask))
2453
},
2454
},
2455
{
2456
name: "EmptyTask",
2457
task: TestOutboundSideEffectTask{},
2458
expectedData: nil,
2459
equalFn: func(t1, t2 any) {
2460
s.IsType(TestOutboundSideEffectTask{}, t1)
2461
s.IsType(TestOutboundSideEffectTask{}, t2)
2462
s.Equal(t1, t2)
2463
},
2464
},
2465
{
2466
name: "StructWithProtoField",
2467
task: &TestPureTask{
2468
Data: payload.Data,
2469
},
2470
expectedData: expectedBlob.GetData(),
2471
equalFn: func(t1, t2 any) {
2472
protorequire.ProtoEqual(s.T(), t1.(*TestPureTask), t2.(*TestPureTask))
2473
},
2474
},
2475
}
2476
2477
for _, tc := range testCases {
2478
s.Run(tc.name, func() {
2479
rt, ok := s.registry.taskFor(tc.task)
2480
s.True(ok)
2481
2482
blob, err := serializeTask(rt, reflect.ValueOf(tc.task))
2483
s.NoError(err)
2484
2485
s.NotNil(blob)
2486
s.Equal(enumspb.ENCODING_TYPE_PROTO3, blob.GetEncodingType())
2487
s.Equal(tc.expectedData, blob.GetData())
2488
2489
deserializedTaskValue, err := deserializeTask(rt, blob)
2490
s.NoError(err)
2491
tc.equalFn(tc.task, deserializedTaskValue.Interface())
2492
})
2493
}
2494
}
2495
2496
func (s *nodeSuite) TestCloseTransaction_Success() {
2497
node := s.testComponentTree()
2498
chasmCtx := NewMutableContext(context.Background(), node)
2499
tc, err := node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2500
s.NoError(err)
2501
tc.(*TestComponent).SubData1 = NewEmptyField[*protoMessageType]()
2502
tc.(*TestComponent).ComponentData = &protoMessageType{CreateRequestId: primitives.NewUUID().String()}
2503
2504
mutations, err := node.CloseTransaction()
2505
s.NoError(err)
2506
s.Len(mutations.UpdatedNodes, 4)
2507
s.Contains(mutations.UpdatedNodes, "", "root component must be in UpdatedNodes")
2508
s.Contains(mutations.UpdatedNodes, "SubComponent1", "SubComponent1 component must be in UpdatedNodes")
2509
s.Contains(mutations.UpdatedNodes, "SubComponent1/SubComponent11", "SubComponent1/SubComponent11 component must be in UpdatedNodes")
2510
s.Contains(mutations.UpdatedNodes, "SubComponent1/SubData11", "SubComponent1/SubData11 component must be in UpdatedNodes")
2511
// SubData1 was never persisted (nil LVT), so no storage delete is needed.
2512
s.Empty(mutations.DeletedNodes)
2513
2514
sc1 := tc.(*TestComponent).SubComponent1.Get(chasmCtx)
2515
s.NotNil(sc1)
2516
2517
mutations, err = node.CloseTransaction()
2518
s.NoError(err)
2519
s.Empty(mutations.UpdatedNodes)
2520
s.Empty(mutations.DeletedNodes)
2521
}
2522
2523
func (s *nodeSuite) TestCloseTransaction_EmptyNode() {
2524
var nilSerializedNodes map[string]*persistencespb.ChasmNode
2525
// Create an empty tree.
2526
node, err := s.newTestTree(nilSerializedNodes)
2527
s.NoError(err)
2528
s.Nil(node.value)
2529
2530
mutations, err := node.CloseTransaction()
2531
s.NoError(err)
2532
s.Empty(mutations.UpdatedNodes, "there should be no updated nodes because tree was initialized with empty serialized nodes")
2533
s.Empty(mutations.DeletedNodes, "there should be no deleted nodes because tree was initialized with empty serialized nodes")
2534
}
2535
2536
func (s *nodeSuite) TestCloseTransaction_LifecycleChange() {
2537
node := s.testComponentTree()
2538
2539
chasmCtx := NewMutableContext(context.Background(), node)
2540
_, err := node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2541
s.NoError(err)
2542
_, err = node.CloseTransaction()
2543
s.NoError(err)
2544
s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING, s.nodeBackend.LastUpdateWorkflowState())
2545
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, s.nodeBackend.LastUpdateWorkflowStatus())
2546
2547
// Test force terminate case
2548
_, err = node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2549
s.NoError(err)
2550
node.terminated = true
2551
_, err = node.CloseTransaction()
2552
s.NoError(err)
2553
s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, s.nodeBackend.LastUpdateWorkflowState())
2554
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, s.nodeBackend.LastUpdateWorkflowStatus())
2555
2556
node.terminated = false
2557
tc, err := node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2558
s.NoError(err)
2559
tc.(*TestComponent).Complete(chasmCtx)
2560
_, err = node.CloseTransaction()
2561
s.NoError(err)
2562
s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, s.nodeBackend.LastUpdateWorkflowState())
2563
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, s.nodeBackend.LastUpdateWorkflowStatus())
2564
2565
tc, err = node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2566
s.NoError(err)
2567
tc.(*TestComponent).Fail(chasmCtx)
2568
_, err = node.CloseTransaction()
2569
s.NoError(err)
2570
s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, s.nodeBackend.LastUpdateWorkflowState())
2571
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_FAILED, s.nodeBackend.LastUpdateWorkflowStatus())
2572
}
2573
2574
func (s *nodeSuite) TestCloseTransaction_ForceUpdateVisibility_RootLifecycleChanged() {
2575
node := s.testComponentTree()
2576
2577
chasmCtx := NewMutableContext(context.Background(), node)
2578
testComponent, err := node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2579
s.NoError(err)
2580
2581
nextTransitionCount := int64(1)
2582
s.nodeBackend.HandleGetCurrentVersion = func() int64 { return 1 }
2583
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTransitionCount }
2584
s.nodeBackend.HandleUpdateWorkflowStateStatus = func(state enumsspb.WorkflowExecutionState, status enumspb.WorkflowExecutionStatus) (bool, error) {
2585
return true, nil
2586
}
2587
2588
// Init visiblity component
2589
testComponent.(*TestComponent).Visibility = NewComponentField(chasmCtx, NewVisibility(chasmCtx))
2590
mutation, err := node.CloseTransaction()
2591
s.NoError(err)
2592
pVisibilityNode, ok := mutation.UpdatedNodes["Visibility"]
2593
s.True(ok)
2594
s.Len(pVisibilityNode.GetMetadata().GetComponentAttributes().SideEffectTasks, 1)
2595
s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING, s.nodeBackend.UpdateCalls[0].State)
2596
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, s.nodeBackend.UpdateCalls[0].Status)
2597
2598
// Change ComponentData which is used as Memo. Even though lifecycle didn't change,
2599
// visibility should be updated because memo changed.
2600
nextTransitionCount = 2
2601
testComponent, err = node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2602
s.NoError(err)
2603
testComponent.(*TestComponent).ComponentData = &protoMessageType{
2604
CreateRequestId: "some-updated-component-data",
2605
}
2606
s.nodeBackend.HandleUpdateWorkflowStateStatus = func(state enumsspb.WorkflowExecutionState, status enumspb.WorkflowExecutionStatus) (bool, error) {
2607
return false, nil
2608
}
2609
mutation, err = node.CloseTransaction()
2610
s.NoError(err)
2611
pVisibilityNode, ok = mutation.UpdatedNodes["Visibility"]
2612
s.True(ok, "visibility should be updated when memo changes")
2613
s.Len(pVisibilityNode.GetMetadata().GetComponentAttributes().SideEffectTasks, 1)
2614
2615
// Close the run, visibility should be force updated
2616
// even if not explicitly updated.
2617
nextTransitionCount = 3
2618
testComponent, err = node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2619
s.NoError(err)
2620
testComponent.(*TestComponent).Complete(chasmCtx)
2621
s.nodeBackend.HandleUpdateWorkflowStateStatus = func(state enumsspb.WorkflowExecutionState, status enumspb.WorkflowExecutionStatus) (bool, error) {
2622
return true, nil
2623
}
2624
mutation, err = node.CloseTransaction()
2625
s.NoError(err)
2626
pVisibilityNode, ok = mutation.UpdatedNodes["Visibility"]
2627
s.True(ok)
2628
s.Len(pVisibilityNode.GetMetadata().GetComponentAttributes().SideEffectTasks, 1)
2629
}
2630
2631
func (s *nodeSuite) TestCloseTransaction_ForceUpdateVisibility_RootSAMemoChanged() {
2632
node := s.testComponentTree()
2633
chasmCtx := NewMutableContext(context.Background(), node)
2634
testComponent, err := node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2635
s.NoError(err)
2636
2637
nextTransitionCount := int64(1)
2638
s.nodeBackend.HandleNextTransitionCount = func() int64 {
2639
return nextTransitionCount
2640
}
2641
2642
// Init visiblity component
2643
testComponent.(*TestComponent).Visibility = NewComponentField(chasmCtx, NewVisibility(chasmCtx))
2644
s.nodeBackend.HandleUpdateWorkflowStateStatus = func(state enumsspb.WorkflowExecutionState, status enumspb.WorkflowExecutionStatus) (bool, error) {
2645
return true, nil
2646
}
2647
mutation, err := node.CloseTransaction()
2648
s.NoError(err)
2649
pVisibilityNode, ok := mutation.UpdatedNodes["Visibility"]
2650
s.True(ok)
2651
s.Len(pVisibilityNode.GetMetadata().GetComponentAttributes().SideEffectTasks, 1)
2652
2653
// Update root component state, which results in a change to the search attributes and memo.
2654
// CHASM framework should automatically detect the change and generate a visibility task.
2655
nextTransitionCount = 2
2656
testComponent, err = node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
2657
s.NoError(err)
2658
testComponent.(*TestComponent).ComponentData = &protoMessageType{
2659
StartTime: timestamppb.Now(),
2660
}
2661
s.nodeBackend.HandleUpdateWorkflowStateStatus = func(state enumsspb.WorkflowExecutionState, status enumspb.WorkflowExecutionStatus) (bool, error) {
2662
return false, nil
2663
}
2664
mutation, err = node.CloseTransaction()
2665
s.NoError(err)
2666
pVisibilityNode, ok = mutation.UpdatedNodes["Visibility"]
2667
s.True(ok)
2668
s.Len(pVisibilityNode.GetMetadata().GetComponentAttributes().SideEffectTasks, 1)
2669
}
2670
2671
func (s *nodeSuite) TestCloseTransaction_CleanupTasksAfterInvalidTask() {
2672
now := s.timeSource.Now().UTC()
2673
2674
task1Attributes := TaskAttributes{}
2675
task1 := &TestPureTask{
2676
Data: []byte("some-random-data"),
2677
}
2678
task2 := &TestPureTask{
2679
Data: []byte("more-random-data"),
2680
}
2681
task1Blob, err := encodeChasmBlob(task1)
2682
s.NoError(err)
2683
task2Blob, err := encodeChasmBlob(task2)
2684
s.NoError(err)
2685
2686
persistenceNodes := map[string]*persistencespb.ChasmNode{
2687
"": {
2688
Metadata: &persistencespb.ChasmNodeMetadata{
2689
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2690
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2691
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
2692
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
2693
TypeId: testComponentTypeID,
2694
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
2695
{
2696
TypeId: testPureTaskTypeID,
2697
ScheduledTime: timestamppb.New(now),
2698
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2699
VersionedTransitionOffset: 3,
2700
Data: task1Blob,
2701
PhysicalTaskStatus: physicalTaskStatusCreated,
2702
},
2703
{
2704
TypeId: testPureTaskTypeID,
2705
ScheduledTime: timestamppb.New(now.Add(1 * time.Minute)),
2706
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2707
VersionedTransitionOffset: 3,
2708
Data: task2Blob,
2709
PhysicalTaskStatus: physicalTaskStatusNone,
2710
},
2711
},
2712
},
2713
},
2714
},
2715
},
2716
}
2717
root, err := s.newTestTree(persistenceNodes)
2718
s.NoError(err)
2719
s.NotNil(root)
2720
2721
s.testLibrary.mockPureTaskHandler.EXPECT().
2722
Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: task1Attributes}), gomock.Eq(task1)).
2723
Return(false, nil).
2724
Times(1)
2725
executed, err := root.ExecutePureTask(s.T().Context(), task1Attributes, task1)
2726
s.NoError(err)
2727
s.False(executed)
2728
s.Equal(valueStateSynced, root.valueState)
2729
2730
nextTransitionCount := int64(1)
2731
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTransitionCount }
2732
2733
s.testLibrary.mockPureTaskHandler.EXPECT().
2734
Validate(gomock.Any(), gomock.Any(), gomock.Any(), protoEq(task1)).
2735
Return(false, nil).
2736
Times(1)
2737
s.testLibrary.mockPureTaskHandler.EXPECT().
2738
Validate(gomock.Any(), gomock.Any(), gomock.Any(), protoEq(task2)).
2739
Return(true, nil).
2740
Times(1)
2741
2742
mutation, err := root.CloseTransaction()
2743
s.NoError(err)
2744
2745
s.Equal(now.Add(1*time.Minute), s.nodeBackend.LastDeletePureTaskCall())
2746
2747
s.Len(mutation.UpdatedNodes, 1)
2748
for _, updatedNode := range mutation.UpdatedNodes {
2749
s.Equal(nextTransitionCount, updatedNode.GetMetadata().GetLastUpdateVersionedTransition().TransitionCount)
2750
}
2751
s.Empty(mutation.DeletedNodes)
2752
2753
componentAttr := root.serializedNode.Metadata.GetComponentAttributes()
2754
s.Len(componentAttr.PureTasks, 1)
2755
s.Equal(testPureTaskTypeID, componentAttr.PureTasks[0].GetTypeId())
2756
s.ProtoEqual(task2Blob, componentAttr.PureTasks[0].GetData())
2757
s.Equal(1, s.nodeBackend.NumTasksAdded()) // physical task is generated for the second pure task
2758
}
2759
2760
func (s *nodeSuite) TestCloseTransaction_InvalidateComponentTasks() {
2761
payload := &commonpb.Payload{
2762
Data: []byte("some-random-data"),
2763
}
2764
taskBlob, err := encodeChasmBlob(payload)
2765
s.NoError(err)
2766
emptyTaskBlob := s.emptyDataBlob()
2767
2768
persistenceNodes := map[string]*persistencespb.ChasmNode{
2769
"": {
2770
Metadata: &persistencespb.ChasmNodeMetadata{
2771
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2772
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2773
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
2774
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
2775
TypeId: testComponentTypeID,
2776
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{
2777
{
2778
TypeId: testSideEffectTaskTypeID,
2779
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2780
VersionedTransitionOffset: 1,
2781
Data: taskBlob,
2782
PhysicalTaskStatus: physicalTaskStatusCreated,
2783
},
2784
{
2785
TypeId: testOutboundSideEffectTaskTypeID,
2786
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2787
VersionedTransitionOffset: 2,
2788
Data: emptyTaskBlob,
2789
PhysicalTaskStatus: physicalTaskStatusCreated,
2790
},
2791
},
2792
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
2793
{
2794
TypeId: testPureTaskTypeID,
2795
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2796
VersionedTransitionOffset: 3,
2797
Data: taskBlob,
2798
PhysicalTaskStatus: physicalTaskStatusCreated,
2799
},
2800
},
2801
},
2802
},
2803
},
2804
},
2805
"SubComponent1": {
2806
Metadata: &persistencespb.ChasmNodeMetadata{
2807
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2808
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2809
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
2810
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
2811
TypeId: testSubComponent1TypeID,
2812
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{
2813
{
2814
TypeId: testSideEffectTaskTypeID,
2815
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2816
VersionedTransitionOffset: 4,
2817
Data: taskBlob,
2818
PhysicalTaskStatus: physicalTaskStatusCreated,
2819
},
2820
},
2821
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
2822
{
2823
TypeId: testPureTaskTypeID,
2824
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2825
VersionedTransitionOffset: 5,
2826
Data: taskBlob,
2827
PhysicalTaskStatus: physicalTaskStatusNone,
2828
},
2829
},
2830
},
2831
},
2832
},
2833
},
2834
}
2835
root, err := s.newTestTree(persistenceNodes)
2836
s.NoError(err)
2837
2838
nextTransitionCount := int64(2)
2839
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTransitionCount }
2840
2841
// The idea is to mark the node as dirty by accessing it with a mutable context.
2842
mutableContext := NewMutableContext(context.Background(), root)
2843
_, err = root.Component(mutableContext, ComponentRef{})
2844
s.NoError(err)
2845
2846
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
2847
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(false, nil).Times(2)
2848
s.testLibrary.mockOutboundSideEffectTaskHandler.EXPECT().
2849
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
2850
s.testLibrary.mockPureTaskHandler.EXPECT().
2851
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(false, nil).Times(2)
2852
2853
mutation, err := root.CloseTransaction()
2854
s.NoError(err)
2855
2856
s.Equal(tasks.MaximumKey.FireTime, s.nodeBackend.LastDeletePureTaskCall())
2857
2858
s.Len(mutation.UpdatedNodes, 2)
2859
for _, updatedNode := range mutation.UpdatedNodes {
2860
s.Equal(nextTransitionCount, updatedNode.GetMetadata().GetLastUpdateVersionedTransition().TransitionCount)
2861
}
2862
s.Empty(mutation.DeletedNodes)
2863
2864
componentAttr := root.serializedNode.Metadata.GetComponentAttributes()
2865
s.Empty(componentAttr.PureTasks)
2866
s.Len(componentAttr.SideEffectTasks, 1)
2867
s.Equal(testOutboundSideEffectTaskTypeID, componentAttr.SideEffectTasks[0].GetTypeId())
2868
2869
componentAttr = root.children["SubComponent1"].serializedNode.Metadata.GetComponentAttributes()
2870
s.Empty(componentAttr.PureTasks)
2871
s.Empty(componentAttr.SideEffectTasks)
2872
}
2873
2874
// TestCloseTransaction_PausedStateInvalidatesTasks verifies that all logical tasks are
2875
// invalidated when a component (or one of its non-detached ancestors) is paused, without
2876
// invoking the task-specific validator.
2877
func (s *nodeSuite) TestCloseTransaction_PausedStateInvalidatesTasks() {
2878
payload := &commonpb.Payload{
2879
Data: []byte("some-random-data"),
2880
}
2881
taskBlob, err := encodeChasmBlob(payload)
2882
s.NoError(err)
2883
2884
makeTask := func(typeID uint32, offset int64) *persistencespb.ChasmComponentAttributes_Task {
2885
return &persistencespb.ChasmComponentAttributes_Task{
2886
TypeId: typeID,
2887
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2888
VersionedTransitionOffset: offset,
2889
Data: taskBlob,
2890
PhysicalTaskStatus: physicalTaskStatusCreated,
2891
}
2892
}
2893
2894
s.Run("paused component invalidates its own tasks without calling task validator", func() {
2895
persistenceNodes := map[string]*persistencespb.ChasmNode{
2896
"": {
2897
Metadata: &persistencespb.ChasmNodeMetadata{
2898
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2899
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2900
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
2901
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
2902
TypeId: testComponentTypeID,
2903
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{makeTask(testSideEffectTaskTypeID, 1)},
2904
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{makeTask(testPureTaskTypeID, 2)},
2905
},
2906
},
2907
},
2908
},
2909
}
2910
root, err := s.newTestTree(persistenceNodes)
2911
s.NoError(err)
2912
2913
nextTransitionCount := int64(2)
2914
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTransitionCount }
2915
2916
// Pause the root component.
2917
mutableContext := NewMutableContext(context.Background(), root)
2918
tc, err := root.Component(mutableContext, ComponentRef{})
2919
s.NoError(err)
2920
tc.(*TestComponent).Pause(mutableContext)
2921
2922
// Task-specific validators must NOT be called - paused state short-circuits them.
2923
// (no EXPECT calls on mock handlers)
2924
2925
mutation, err := root.CloseTransaction()
2926
s.NoError(err)
2927
2928
componentAttr := root.serializedNode.Metadata.GetComponentAttributes()
2929
s.Empty(componentAttr.SideEffectTasks, "paused component should have no side-effect tasks")
2930
s.Empty(componentAttr.PureTasks, "paused component should have no pure tasks")
2931
2932
// Node must be marked updated so the invalidation is persisted.
2933
s.Len(mutation.UpdatedNodes, 1)
2934
})
2935
2936
s.Run("paused parent invalidates non-detached sub-component tasks", func() {
2937
persistenceNodes := map[string]*persistencespb.ChasmNode{
2938
"": {
2939
Metadata: &persistencespb.ChasmNodeMetadata{
2940
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2941
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2942
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
2943
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
2944
TypeId: testComponentTypeID,
2945
},
2946
},
2947
},
2948
},
2949
"SubComponent1": {
2950
Metadata: &persistencespb.ChasmNodeMetadata{
2951
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2952
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2953
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
2954
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
2955
TypeId: testSubComponent1TypeID,
2956
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{makeTask(testSideEffectTaskTypeID, 1)},
2957
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{makeTask(testPureTaskTypeID, 2)},
2958
},
2959
},
2960
},
2961
},
2962
}
2963
root, err := s.newTestTree(persistenceNodes)
2964
s.NoError(err)
2965
2966
nextTransitionCount := int64(2)
2967
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTransitionCount }
2968
2969
// Pause the root - its non-detached sub-component's tasks should also be invalidated.
2970
mutableContext := NewMutableContext(context.Background(), root)
2971
tc, err := root.Component(mutableContext, ComponentRef{})
2972
s.NoError(err)
2973
tc.(*TestComponent).Pause(mutableContext)
2974
2975
mutation, err := root.CloseTransaction()
2976
s.NoError(err)
2977
2978
subAttr := root.children["SubComponent1"].serializedNode.Metadata.GetComponentAttributes()
2979
s.Empty(subAttr.SideEffectTasks, "non-detached sub-component tasks should be invalidated when parent is paused")
2980
s.Empty(subAttr.PureTasks)
2981
s.Len(mutation.UpdatedNodes, 2) // root (paused) + SubComponent1 (task cleanup)
2982
})
2983
2984
s.Run("detached sub-component tasks are NOT invalidated by parent pause", func() {
2985
persistenceNodes := map[string]*persistencespb.ChasmNode{
2986
"": {
2987
Metadata: &persistencespb.ChasmNodeMetadata{
2988
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2989
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
2990
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
2991
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
2992
TypeId: testComponentTypeID,
2993
},
2994
},
2995
},
2996
},
2997
"SubComponent1": {
2998
Metadata: &persistencespb.ChasmNodeMetadata{
2999
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3000
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3001
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3002
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3003
TypeId: testSubComponent1TypeID,
3004
Detached: true,
3005
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{makeTask(testSideEffectTaskTypeID, 1)},
3006
},
3007
},
3008
},
3009
},
3010
}
3011
root, err := s.newTestTree(persistenceNodes)
3012
s.NoError(err)
3013
3014
nextTransitionCount := int64(2)
3015
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTransitionCount }
3016
3017
// Pause the root.
3018
mutableContext := NewMutableContext(context.Background(), root)
3019
tc, err := root.Component(mutableContext, ComponentRef{})
3020
s.NoError(err)
3021
tc.(*TestComponent).Pause(mutableContext)
3022
3023
// The detached sub-component's validator IS called (it decides independently).
3024
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
3025
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
3026
3027
mutation, err := root.CloseTransaction()
3028
s.NoError(err)
3029
3030
subAttr := root.children["SubComponent1"].serializedNode.Metadata.GetComponentAttributes()
3031
s.Len(subAttr.SideEffectTasks, 1, "detached sub-component tasks should survive parent pause")
3032
_ = mutation
3033
})
3034
3035
s.Run("write access accepted on paused component", func() {
3036
// Requirement: for now accept chasm engine requests on paused component.
3037
root, err := s.newTestTree(testComponentSerializedNodes())
3038
s.NoError(err)
3039
3040
ctx := NewContext(
3041
newContextWithOperationIntent(context.Background(), OperationIntentProgress),
3042
root,
3043
)
3044
3045
// Pause the root.
3046
err = root.prepareComponentValue(ctx)
3047
s.NoError(err)
3048
root.value.(*TestComponent).Pause(NewMutableContext(context.Background(), root))
3049
3050
// validateAccess should still succeed - paused does NOT block writes.
3051
subNode, ok := root.findNode([]string{"SubComponent1"})
3052
s.True(ok)
3053
err = subNode.validateAccess(ctx, false)
3054
s.NoError(err, "write access to sub-component of paused parent should be accepted")
3055
})
3056
}
3057
3058
// TestCloseTransaction_TaskValidationSubtreeDirty verifies that task validation is
3059
// skipped for component nodes whose entire lineage is clean, and runs for nodes
3060
// whose own subtree or an ancestor is dirty.
3061
func (s *nodeSuite) TestCloseTransaction_TaskValidationSubtreeDirty() {
3062
payload := &commonpb.Payload{Data: []byte("some-random-data")}
3063
taskBlob, err := encodeChasmBlob(payload)
3064
s.NoError(err)
3065
3066
makeTask := func(typeID uint32, offset int64) *persistencespb.ChasmComponentAttributes_Task {
3067
return &persistencespb.ChasmComponentAttributes_Task{
3068
TypeId: typeID,
3069
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3070
VersionedTransitionOffset: offset,
3071
Data: taskBlob,
3072
PhysicalTaskStatus: physicalTaskStatusCreated,
3073
}
3074
}
3075
3076
// Tree shape: root (testComponent) with two sibling children SubComponent1 and SubComponent2.
3077
baseNodes := func() map[string]*persistencespb.ChasmNode {
3078
return map[string]*persistencespb.ChasmNode{
3079
"": {
3080
Metadata: &persistencespb.ChasmNodeMetadata{
3081
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3082
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3083
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3084
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3085
TypeId: testComponentTypeID,
3086
},
3087
},
3088
},
3089
},
3090
"SubComponent1": {
3091
Metadata: &persistencespb.ChasmNodeMetadata{
3092
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3093
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3094
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3095
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3096
TypeId: testSubComponent1TypeID,
3097
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{makeTask(testSideEffectTaskTypeID, 1)},
3098
},
3099
},
3100
},
3101
},
3102
"SubComponent2": {
3103
Metadata: &persistencespb.ChasmNodeMetadata{
3104
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3105
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3106
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3107
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3108
TypeId: testSubComponent2TypeID,
3109
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{makeTask(testSideEffectTaskTypeID, 1)},
3110
},
3111
},
3112
},
3113
},
3114
}
3115
}
3116
3117
s.Run("unrelated sibling subtree tasks are not validated", func() {
3118
// Dirty SubComponent1. SubComponent2 is untouched and in an unrelated subtree,
3119
// so its task validator must not be called.
3120
root, err := s.newTestTree(baseNodes())
3121
s.NoError(err)
3122
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
3123
3124
mutableCtx := NewMutableContext(context.Background(), root)
3125
sc1Ref := ComponentRef{componentPath: []string{"SubComponent1"}}
3126
sc1, err := root.Component(mutableCtx, sc1Ref)
3127
s.NoError(err)
3128
// Mutate SubComponent1 to make it dirty.
3129
sc1.(*TestSubComponent1).SubComponent1Data = &protoMessageType{}
3130
3131
// SubComponent1's task validator is called (it's dirty).
3132
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
3133
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
3134
3135
// SubComponent2's validator must NOT be called - its subtree is clean.
3136
// (no EXPECT on mockSideEffectTaskHandler for SubComponent2)
3137
3138
_, err = root.CloseTransaction()
3139
s.NoError(err)
3140
3141
sc2Attr := root.children["SubComponent2"].serializedNode.Metadata.GetComponentAttributes()
3142
s.Len(sc2Attr.SideEffectTasks, 1, "unrelated subtree tasks should be untouched")
3143
})
3144
3145
s.Run("dirty ancestor causes descendant tasks to be validated", func() {
3146
// Dirty the root. SubComponent1 is a descendant and must have its tasks validated
3147
// because a parent closing/pausing affects descendant task validity.
3148
root, err := s.newTestTree(baseNodes())
3149
s.NoError(err)
3150
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
3151
3152
mutableCtx := NewMutableContext(context.Background(), root)
3153
tc, err := root.Component(mutableCtx, ComponentRef{})
3154
s.NoError(err)
3155
tc.(*TestComponent).Pause(mutableCtx)
3156
3157
// Both sub-components' tasks are invalidated by the paused ancestor
3158
// (validateAccess short-circuits before calling task validators).
3159
_, err = root.CloseTransaction()
3160
s.NoError(err)
3161
3162
sc1Attr := root.children["SubComponent1"].serializedNode.Metadata.GetComponentAttributes()
3163
s.Empty(sc1Attr.SideEffectTasks, "descendant tasks should be invalidated when ancestor is paused")
3164
sc2Attr := root.children["SubComponent2"].serializedNode.Metadata.GetComponentAttributes()
3165
s.Empty(sc2Attr.SideEffectTasks, "descendant tasks should be invalidated when ancestor is paused")
3166
})
3167
3168
s.Run("dirty descendant causes ancestor tasks to be validated", func() {
3169
// Give the root component a task, then dirty a child. The root's task validator
3170
// must be called because a descendant's state can affect ancestor task validity.
3171
nodes := baseNodes()
3172
nodes[""].Metadata.GetComponentAttributes().SideEffectTasks = []*persistencespb.ChasmComponentAttributes_Task{
3173
makeTask(testSideEffectTaskTypeID, 1),
3174
}
3175
root, err := s.newTestTree(nodes)
3176
s.NoError(err)
3177
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
3178
3179
mutableCtx := NewMutableContext(context.Background(), root)
3180
sc1Ref := ComponentRef{componentPath: []string{"SubComponent1"}}
3181
sc1, err := root.Component(mutableCtx, sc1Ref)
3182
s.NoError(err)
3183
sc1.(*TestSubComponent1).SubComponent1Data = &protoMessageType{}
3184
3185
// Root's validator is called because its subtree (SubComponent1) is dirty.
3186
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
3187
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
3188
// SubComponent1's validator is also called (it's dirty itself).
3189
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
3190
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
3191
3192
_, err = root.CloseTransaction()
3193
s.NoError(err)
3194
})
3195
3196
s.Run("subtreeIsDirty resets after CloseTransaction", func() {
3197
root, err := s.newTestTree(baseNodes())
3198
s.NoError(err)
3199
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
3200
3201
mutableCtx := NewMutableContext(context.Background(), root)
3202
sc1Ref := ComponentRef{componentPath: []string{"SubComponent1"}}
3203
sc1, err := root.Component(mutableCtx, sc1Ref)
3204
s.NoError(err)
3205
sc1.(*TestSubComponent1).SubComponent1Data = &protoMessageType{}
3206
3207
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
3208
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
3209
3210
_, err = root.CloseTransaction()
3211
s.NoError(err)
3212
3213
// After CloseTransaction, subtreeIsDirty must be reset on all nodes.
3214
for _, node := range root.andAllChildren() {
3215
s.False(node.subtreeIsDirty, "subtreeIsDirty must be reset after CloseTransaction")
3216
}
3217
})
3218
3219
s.Run("ExecutePureTask with no state mutations still cleans up the executed task", func() {
3220
nodes := map[string]*persistencespb.ChasmNode{
3221
"": {
3222
Metadata: &persistencespb.ChasmNodeMetadata{
3223
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3224
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3225
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3226
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3227
TypeId: testComponentTypeID,
3228
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{makeTask(testPureTaskTypeID, 1)},
3229
},
3230
},
3231
},
3232
},
3233
}
3234
root, err := s.newTestTree(nodes)
3235
s.NoError(err)
3236
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
3237
3238
pureTask := &TestPureTask{Data: []byte("some-data")}
3239
3240
// Validator returns invalid: once in ExecutePureTask's own check, and once more
3241
// during CloseTransaction's task cleanup pass (triggered by markSubtreeDirty).
3242
s.testLibrary.mockPureTaskHandler.EXPECT().
3243
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(false, nil).Times(2)
3244
3245
executed, err := root.ExecutePureTask(context.Background(), TaskAttributes{}, pureTask)
3246
s.NoError(err)
3247
s.False(executed)
3248
s.True(root.subtreeIsDirty, "ExecutePureTask must mark subtreeIsDirty even when task is invalid")
3249
3250
_, err = root.CloseTransaction()
3251
s.NoError(err)
3252
3253
componentAttr := root.serializedNode.Metadata.GetComponentAttributes()
3254
s.Empty(componentAttr.PureTasks, "invalid pure task should be cleaned up after CloseTransaction")
3255
})
3256
}
3257
3258
func (s *nodeSuite) TestCloseTransaction_LifecycleChange_PausedRootKeepsRunning() {
3259
// When the root component is paused, the execution state should remain RUNNING
3260
// because paused is an OPEN lifecycle state.
3261
node := s.testComponentTree()
3262
3263
chasmCtx := NewMutableContext(context.Background(), node)
3264
rootComp, err := node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
3265
s.NoError(err)
3266
rootComp.(*TestComponent).Pause(chasmCtx)
3267
3268
_, err = node.CloseTransaction()
3269
s.NoError(err)
3270
s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING, s.nodeBackend.LastUpdateWorkflowState())
3271
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, s.nodeBackend.LastUpdateWorkflowStatus())
3272
}
3273
3274
func (s *nodeSuite) TestCloseTransaction_NewComponentTasks() {
3275
persistenceNodes := map[string]*persistencespb.ChasmNode{
3276
"": {
3277
Metadata: &persistencespb.ChasmNodeMetadata{
3278
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3279
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3280
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3281
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3282
TypeId: testComponentTypeID,
3283
},
3284
},
3285
},
3286
},
3287
"SubComponent1": {
3288
Metadata: &persistencespb.ChasmNodeMetadata{
3289
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3290
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3291
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3292
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3293
TypeId: testSubComponent1TypeID,
3294
},
3295
},
3296
},
3297
},
3298
"SubComponent2": {
3299
Metadata: &persistencespb.ChasmNodeMetadata{
3300
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3301
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3302
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3303
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3304
TypeId: testSubComponent2TypeID,
3305
},
3306
},
3307
},
3308
},
3309
}
3310
3311
s.nodeBackend.HandleNextTransitionCount = func() int64 {
3312
return 2
3313
}
3314
3315
root, err := s.newTestTree(persistenceNodes)
3316
s.NoError(err)
3317
3318
mutableContext := NewMutableContext(context.Background(), root)
3319
c, err := root.Component(mutableContext, ComponentRef{})
3320
s.NoError(err)
3321
3322
// Add a valid side effect task.
3323
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
3324
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
3325
testComponent := c.(*TestComponent)
3326
mutableContext.AddTask(testComponent, TaskAttributes{}, &TestSideEffectTask{
3327
Data: []byte("some-random-data"),
3328
})
3329
3330
// Add an invalid outbound side effect task.
3331
// the invalid task should not be created.
3332
s.testLibrary.mockOutboundSideEffectTaskHandler.EXPECT().
3333
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(false, nil).Times(1)
3334
mutableContext.AddTask(
3335
testComponent,
3336
TaskAttributes{Destination: "destination"},
3337
TestOutboundSideEffectTask{},
3338
)
3339
3340
// Add a valid pure task.
3341
s.testLibrary.mockPureTaskHandler.EXPECT().
3342
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
3343
mutableContext.AddTask(
3344
testComponent,
3345
TaskAttributes{ScheduledTime: s.timeSource.Now()},
3346
&TestPureTask{
3347
Data: []byte("valid-pure-task"),
3348
},
3349
)
3350
3351
// Add an invalid pure task.
3352
// the invalid task should not be created.
3353
s.testLibrary.mockPureTaskHandler.EXPECT().
3354
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(false, nil).Times(1)
3355
mutableContext.AddTask(
3356
testComponent,
3357
TaskAttributes{ScheduledTime: s.timeSource.Now()},
3358
&TestPureTask{
3359
Data: []byte("invalid-pure-task"),
3360
},
3361
)
3362
3363
// Add a valid outbound side effect task to a sub-component.
3364
s.testLibrary.mockOutboundSideEffectTaskHandler.EXPECT().
3365
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
3366
subComponent2 := testComponent.SubComponent2.Get(mutableContext)
3367
mutableContext.AddTask(
3368
subComponent2,
3369
TaskAttributes{Destination: "destination"},
3370
TestOutboundSideEffectTask{},
3371
)
3372
3373
mutation, err := root.CloseTransaction()
3374
s.NoError(err)
3375
3376
s.Equal(s.timeSource.Now().UTC(), s.nodeBackend.LastDeletePureTaskCall())
3377
3378
rootAttr := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
3379
s.Len(rootAttr.SideEffectTasks, 1) // Only one valid side effect task.
3380
newSideEffectTask := rootAttr.SideEffectTasks[0]
3381
newSideEffectTask.Data = nil // This is tested by TestSerializeTask()
3382
s.ProtoEqual(&persistencespb.ChasmComponentAttributes_Task{
3383
TypeId: testSideEffectTaskTypeID,
3384
ScheduledTime: timestamppb.New(time.Time{}),
3385
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3386
VersionedTransitionOffset: 1,
3387
PhysicalTaskStatus: physicalTaskStatusCreated,
3388
}, newSideEffectTask)
3389
s.Len(s.nodeBackend.TasksByCategory[tasks.CategoryTransfer], 1)
3390
chasmTask := s.nodeBackend.TasksByCategory[tasks.CategoryTransfer][0].(*tasks.ChasmTask)
3391
s.ProtoEqual(&persistencespb.ChasmTaskInfo{
3392
ComponentInitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3393
ComponentLastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3394
Path: rootPath,
3395
TypeId: testSideEffectTaskTypeID,
3396
Data: chasmTask.Info.GetData(), // This is tested by TestSerializeTask()
3397
ArchetypeId: testComponentTypeID,
3398
TaskVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3399
TaskVersionedTransitionOffset: 1,
3400
}, chasmTask.Info)
3401
3402
s.Len(rootAttr.PureTasks, 1) // Only one valid side effect task.
3403
newPureTask := rootAttr.PureTasks[0]
3404
newPureTask.Data = nil // This is tested by TestSerializeTask()
3405
s.ProtoEqual(&persistencespb.ChasmComponentAttributes_Task{
3406
TypeId: testPureTaskTypeID,
3407
ScheduledTime: timestamppb.New(s.timeSource.Now()),
3408
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3409
VersionedTransitionOffset: 2,
3410
PhysicalTaskStatus: physicalTaskStatusCreated,
3411
}, newPureTask)
3412
s.Len(s.nodeBackend.TasksByCategory[tasks.CategoryTimer], 1)
3413
chasmPureTask := s.nodeBackend.TasksByCategory[tasks.CategoryTimer][0].(*tasks.ChasmTaskPure)
3414
s.Equal(tasks.CategoryTimer, chasmPureTask.GetCategory())
3415
s.True(chasmPureTask.VisibilityTimestamp.Equal(s.timeSource.Now()))
3416
3417
subComponent2Attr := mutation.UpdatedNodes["SubComponent2"].GetMetadata().GetComponentAttributes()
3418
newOutboundSideEffectTask := subComponent2Attr.SideEffectTasks[0]
3419
newOutboundSideEffectTask.Data = nil // This is tested by TestSerializeTask()
3420
s.ProtoEqual(&persistencespb.ChasmComponentAttributes_Task{
3421
TypeId: testOutboundSideEffectTaskTypeID,
3422
Destination: "destination",
3423
ScheduledTime: timestamppb.New(time.Time{}),
3424
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3425
VersionedTransitionOffset: 3,
3426
PhysicalTaskStatus: physicalTaskStatusCreated,
3427
}, newOutboundSideEffectTask)
3428
s.Len(s.nodeBackend.TasksByCategory[tasks.CategoryOutbound], 1)
3429
chasmTask = s.nodeBackend.TasksByCategory[tasks.CategoryOutbound][0].(*tasks.ChasmTask)
3430
s.ProtoEqual(&persistencespb.ChasmTaskInfo{
3431
ComponentInitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3432
ComponentLastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3433
Path: []string{"SubComponent2"},
3434
TypeId: testOutboundSideEffectTaskTypeID,
3435
Data: chasmTask.Info.GetData(), // This is tested by TestSerializeTask()
3436
ArchetypeId: testComponentTypeID,
3437
TaskVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3438
TaskVersionedTransitionOffset: 3,
3439
}, chasmTask.Info)
3440
}
3441
3442
func (s *nodeSuite) TestCloseTransaction_ApplyMutation_SideEffectTasks() {
3443
persistenceNodes := map[string]*persistencespb.ChasmNode{
3444
"": {
3445
Metadata: &persistencespb.ChasmNodeMetadata{
3446
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3447
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3448
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3449
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3450
TypeId: testComponentTypeID,
3451
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{
3452
{
3453
TypeId: testSideEffectTaskTypeID,
3454
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3455
VersionedTransitionOffset: 1,
3456
PhysicalTaskStatus: physicalTaskStatusCreated,
3457
},
3458
},
3459
},
3460
},
3461
},
3462
},
3463
}
3464
3465
incomingMutation := NodesMutation{
3466
UpdatedNodes: map[string]*persistencespb.ChasmNode{
3467
"": {
3468
Metadata: &persistencespb.ChasmNodeMetadata{
3469
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3470
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3471
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3472
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3473
TypeId: testComponentTypeID,
3474
SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{
3475
{
3476
TypeId: testSideEffectTaskTypeID,
3477
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3478
VersionedTransitionOffset: 1,
3479
PhysicalTaskStatus: physicalTaskStatusCreated,
3480
},
3481
{
3482
TypeId: testSideEffectTaskTypeID,
3483
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3484
VersionedTransitionOffset: 1,
3485
PhysicalTaskStatus: physicalTaskStatusNone,
3486
},
3487
{
3488
TypeId: testSideEffectTaskTypeID,
3489
Destination: "destination",
3490
ScheduledTime: timestamppb.New(TaskScheduledTimeImmediate),
3491
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3492
VersionedTransitionOffset: 2,
3493
PhysicalTaskStatus: physicalTaskStatusNone,
3494
},
3495
{
3496
TypeId: testSideEffectTaskTypeID,
3497
ScheduledTime: timestamppb.New(s.timeSource.Now().Add(time.Minute)),
3498
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3499
VersionedTransitionOffset: 3,
3500
PhysicalTaskStatus: physicalTaskStatusNone,
3501
},
3502
},
3503
},
3504
},
3505
},
3506
},
3507
},
3508
}
3509
3510
root, err := s.newTestTree(persistenceNodes)
3511
s.NoError(err)
3512
3513
err = root.ApplyMutation(incomingMutation)
3514
s.NoError(err)
3515
3516
expectedCategories := []tasks.Category{tasks.CategoryTimer, tasks.CategoryOutbound, tasks.CategoryTransfer}
3517
_, err = root.CloseTransaction()
3518
for _, category := range expectedCategories {
3519
for _, task := range s.nodeBackend.TasksByCategory[category] {
3520
s.IsType(&tasks.ChasmTask{}, task)
3521
s.Equal(category, task.GetCategory())
3522
}
3523
}
3524
3525
s.NoError(err)
3526
}
3527
3528
func (s *nodeSuite) TestCloseTransaction_ApplyMutation_PureTasks() {
3529
now := s.timeSource.Now().UTC()
3530
persistenceNodes := map[string]*persistencespb.ChasmNode{
3531
"": {
3532
Metadata: &persistencespb.ChasmNodeMetadata{
3533
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3534
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3535
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3536
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3537
TypeId: testComponentTypeID,
3538
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
3539
{
3540
TypeId: testPureTaskTypeID,
3541
ScheduledTime: timestamppb.New(now.Add(time.Second)),
3542
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3543
VersionedTransitionOffset: 1,
3544
PhysicalTaskStatus: physicalTaskStatusCreated,
3545
},
3546
},
3547
},
3548
},
3549
},
3550
},
3551
"SubComponent1": {
3552
Metadata: &persistencespb.ChasmNodeMetadata{
3553
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3554
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3555
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3556
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3557
TypeId: testSubComponent1TypeID,
3558
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
3559
{
3560
TypeId: testPureTaskTypeID,
3561
ScheduledTime: timestamppb.New(now.Add(time.Minute)),
3562
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3563
VersionedTransitionOffset: 2,
3564
PhysicalTaskStatus: physicalTaskStatusNone,
3565
},
3566
},
3567
},
3568
},
3569
},
3570
},
3571
}
3572
3573
incomingMutation := NodesMutation{
3574
UpdatedNodes: map[string]*persistencespb.ChasmNode{
3575
"": {
3576
Metadata: &persistencespb.ChasmNodeMetadata{
3577
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3578
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3579
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3580
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3581
TypeId: testComponentTypeID,
3582
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
3583
{
3584
TypeId: testPureTaskTypeID,
3585
ScheduledTime: timestamppb.New(now.Add(2 * time.Minute)),
3586
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 2},
3587
VersionedTransitionOffset: 1,
3588
PhysicalTaskStatus: physicalTaskStatusNone,
3589
},
3590
},
3591
},
3592
},
3593
},
3594
},
3595
},
3596
}
3597
3598
root, err := s.newTestTree(persistenceNodes)
3599
s.NoError(err)
3600
3601
err = root.ApplyMutation(incomingMutation)
3602
s.NoError(err)
3603
3604
mutation, err := root.CloseTransaction()
3605
s.NoError(err)
3606
3607
s.Equal(now.Add(time.Minute), s.nodeBackend.LastDeletePureTaskCall())
3608
3609
// Although only root is mutated in ApplyMutation, we generated a pure task for the child node,
3610
// and need to persist that as well.
3611
s.Len(mutation.UpdatedNodes, 2)
3612
3613
s.Len(s.nodeBackend.TasksByCategory[tasks.CategoryTimer], 1)
3614
task := s.nodeBackend.TasksByCategory[tasks.CategoryTimer][0]
3615
s.IsType(&tasks.ChasmTaskPure{}, task)
3616
s.True(now.Add(time.Minute).Equal(task.GetKey().FireTime))
3617
}
3618
3619
func (s *nodeSuite) TestTerminate() {
3620
node := s.testComponentTree()
3621
3622
// First closeTransaction once to make the tree clean.
3623
_, err := node.CloseTransaction()
3624
s.NoError(err)
3625
3626
s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING, s.nodeBackend.LastUpdateWorkflowState())
3627
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, s.nodeBackend.LastUpdateWorkflowStatus())
3628
3629
// Then terminate the node and verify only that node will be in the mutation.
3630
err = node.Terminate(TerminateComponentRequest{})
3631
s.NoError(err)
3632
s.True(node.terminated)
3633
3634
mutations, err := node.CloseTransaction()
3635
s.NoError(err)
3636
s.Len(mutations.UpdatedNodes, 1)
3637
s.Empty(mutations.DeletedNodes)
3638
s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, s.nodeBackend.LastUpdateWorkflowState())
3639
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, s.nodeBackend.LastUpdateWorkflowStatus())
3640
3641
// Test updating a terminated node will NOT change the state & status in mutable state.
3642
// Here we simulate mutable state reload case since the terminate flag is not persisted.
3643
s.nodeBackend.HandleGetExecutionState = func() *persistencespb.WorkflowExecutionState {
3644
return &persistencespb.WorkflowExecutionState{
3645
State: enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
3646
Status: enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED,
3647
}
3648
}
3649
3650
snapshot := node.Snapshot(nil)
3651
node, err = s.newTestTree(snapshot.Nodes)
3652
s.NoError(err)
3653
3654
mutableContext := NewMutableContext(context.Background(), node)
3655
_, err = node.Component(mutableContext, ComponentRef{})
3656
s.NoError(err)
3657
3658
mutations, err = node.CloseTransaction()
3659
s.NoError(err)
3660
s.Empty(mutations.UpdatedNodes)
3661
s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, s.nodeBackend.LastUpdateWorkflowState())
3662
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, s.nodeBackend.LastUpdateWorkflowStatus())
3663
}
3664
3665
func (s *nodeSuite) preorderAndAssertParent(
3666
n *Node,
3667
parent *Node,
3668
) []*persistencespb.ChasmNode {
3669
s.Equal(parent, n.parent)
3670
3671
var nodes []*persistencespb.ChasmNode
3672
nodes = append(nodes, n.serializedNode)
3673
3674
childNames := make([]string, 0, len(n.children))
3675
for childName := range n.children {
3676
childNames = append(childNames, childName)
3677
}
3678
sort.Strings(childNames)
3679
3680
for _, childName := range childNames {
3681
nodes = append(nodes, s.preorderAndAssertParent(n.children[childName], n)...)
3682
}
3683
3684
return nodes
3685
}
3686
3687
type testNodePathEncoder struct{}
3688
3689
var _ NodePathEncoder = (*testNodePathEncoder)(nil)
3690
3691
func (e *testNodePathEncoder) Encode(
3692
_ *Node,
3693
path []string,
3694
) (string, error) {
3695
return strings.Join(path, "/"), nil
3696
}
3697
3698
func (e *testNodePathEncoder) Decode(
3699
encodedPath string,
3700
) ([]string, error) {
3701
if encodedPath == "" {
3702
return rootPath, nil
3703
}
3704
return strings.Split(encodedPath, "/"), nil
3705
}
3706
3707
func (s *nodeSuite) nodeBase() *nodeBase {
3708
return &nodeBase{
3709
registry: s.registry,
3710
timeSource: s.timeSource,
3711
backend: s.nodeBackend,
3712
pathEncoder: s.nodePathEncoder,
3713
3714
mutation: NodesMutation{
3715
UpdatedNodes: make(map[string]*persistencespb.ChasmNode),
3716
DeletedNodes: make(map[string]struct{}),
3717
},
3718
systemMutation: NodesMutation{
3719
UpdatedNodes: make(map[string]*persistencespb.ChasmNode),
3720
DeletedNodes: make(map[string]struct{}),
3721
},
3722
newTasks: make(map[any][]taskWithAttributes),
3723
taskValueCache: make(map[*commonpb.DataBlob]reflect.Value),
3724
}
3725
}
3726
3727
// Helper method to create a test tree for TestComponent.
3728
func (s *nodeSuite) testComponentTree() *Node {
3729
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 1 }
3730
s.nodeBackend.HandleGetCurrentVersion = func() int64 { return 1 }
3731
3732
var nilSerializedNodes map[string]*persistencespb.ChasmNode
3733
// Create an empty tree.
3734
node, err := s.newTestTree(nilSerializedNodes)
3735
s.NoError(err)
3736
s.Nil(node.value)
3737
3738
tc := &TestComponent{}
3739
setTestComponentFields(tc, s.nodeBackend)
3740
err = node.SetRootComponent(tc)
3741
s.False(node.needsPointerResolution)
3742
s.NoError(err)
3743
s.Empty(node.mutation.DeletedNodes)
3744
3745
return node // maybe tc too
3746
}
3747
3748
func (s *nodeSuite) TestContextNowStableWithinContext() {
3749
root := s.testComponentTree()
3750
3751
startTime := time.Date(2026, 1, 1, 1, 0, 0, 0, time.UTC)
3752
updatedTime := startTime.Add(time.Minute)
3753
laterTime := updatedTime.Add(time.Minute)
3754
finalTime := laterTime.Add(time.Minute)
3755
3756
s.timeSource.Update(startTime)
3757
3758
mutableContext := NewMutableContext(context.Background(), root)
3759
s.timeSource.Update(updatedTime)
3760
3761
component, err := root.Component(mutableContext, ComponentRef{})
3762
s.NoError(err)
3763
testComponent := component.(*TestComponent)
3764
3765
s.Equal(startTime, mutableContext.Now(component))
3766
s.Equal(startTime, mutableContext.Now(component))
3767
3768
childComponent := testComponent.SubComponent1.Get(mutableContext)
3769
s.Equal(startTime, mutableContext.Now(childComponent))
3770
3771
contextWithValue := ContextWithValue(mutableContext, "test-key", "test-value")
3772
s.Equal("test-value", contextWithValue.Value("test-key"))
3773
s.Equal(startTime, contextWithValue.Now(component))
3774
3775
s.timeSource.Update(laterTime)
3776
s.Equal(startTime, contextWithValue.Now(component))
3777
s.Equal(laterTime, NewMutableContext(context.Background(), root).Now(component))
3778
3779
immutableContext := NewContext(context.Background(), root)
3780
s.Equal(laterTime, immutableContext.Now(component))
3781
3782
s.timeSource.Update(finalTime)
3783
s.Equal(laterTime, immutableContext.Now(component))
3784
}
3785
3786
func (s *nodeSuite) TestExecuteImmediatePureTask() {
3787
root := s.testComponentTree()
3788
3789
mutations, err := root.CloseTransaction()
3790
s.NoError(err)
3791
3792
// Start a clean transaction.
3793
3794
mutableContext := NewMutableContext(context.Background(), root)
3795
component, err := root.Component(mutableContext, ComponentRef{})
3796
s.NoError(err)
3797
testComponent := component.(*TestComponent)
3798
3799
taskAttributes := TaskAttributes{ScheduledTime: TaskScheduledTimeImmediate}
3800
mutableContext.AddTask(
3801
testComponent,
3802
taskAttributes,
3803
&TestPureTask{
3804
Data: []byte("root-task-payload"),
3805
},
3806
)
3807
3808
sc1 := testComponent.SubComponent1.Get(mutableContext)
3809
3810
mutableContext.AddTask(
3811
sc1,
3812
taskAttributes,
3813
&TestPureTask{
3814
Data: []byte("sc1-task-payload"),
3815
},
3816
)
3817
3818
// One valid task, one invalid task
3819
s.testLibrary.mockPureTaskHandler.EXPECT().
3820
Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: taskAttributes}), gomock.Any()).Return(false, nil).Times(1)
3821
s.testLibrary.mockPureTaskHandler.EXPECT().
3822
Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: taskAttributes}), gomock.Any()).Return(true, nil).Times(1)
3823
s.testLibrary.mockPureTaskHandler.EXPECT().
3824
Execute(
3825
gomock.AssignableToTypeOf(&mutableCtx{}),
3826
gomock.Any(),
3827
gomock.Eq(taskAttributes),
3828
gomock.Any(),
3829
).Return(nil).Times(1)
3830
3831
mutations, err = root.CloseTransaction()
3832
s.NoError(err)
3833
s.Empty(mutations.UpdatedNodes)
3834
s.Empty(mutations.DeletedNodes)
3835
3836
// immedidate pure tasks will be executed inline and no physical chasm pure task will be generated.
3837
s.Equal(tasks.MaximumKey.FireTime, s.nodeBackend.LastDeletePureTaskCall())
3838
}
3839
3840
func (s *nodeSuite) TestImmediatePureTaskNowStableWithinTaskOnly() {
3841
root := s.testComponentTree()
3842
3843
_, err := root.CloseTransaction()
3844
s.NoError(err)
3845
3846
taskStartTime := time.Date(2026, 1, 1, 2, 0, 0, 0, time.UTC)
3847
nextTaskTime := taskStartTime.Add(time.Minute)
3848
s.timeSource.Update(taskStartTime)
3849
3850
mutableContext := NewMutableContext(context.Background(), root)
3851
component, err := root.Component(mutableContext, ComponentRef{})
3852
s.NoError(err)
3853
3854
taskAttributes := TaskAttributes{ScheduledTime: TaskScheduledTimeImmediate}
3855
mutableContext.AddTask(
3856
component,
3857
taskAttributes,
3858
&TestPureTask{},
3859
)
3860
mutableContext.AddTask(
3861
component,
3862
taskAttributes,
3863
&TestPureTask{},
3864
)
3865
3866
s.testLibrary.mockPureTaskHandler.EXPECT().
3867
Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: taskAttributes}), gomock.Any()).Return(true, nil).Times(2)
3868
3869
var observedTimes []time.Time
3870
s.testLibrary.mockPureTaskHandler.EXPECT().
3871
Execute(
3872
gomock.AssignableToTypeOf(&mutableCtx{}),
3873
gomock.AssignableToTypeOf(&TestComponent{}),
3874
gomock.Eq(taskAttributes),
3875
gomock.Any(),
3876
).
3877
DoAndReturn(func(ctx MutableContext, component any, _ TaskAttributes, _ *TestPureTask) error {
3878
chasmComponent := component.(Component)
3879
firstNow := ctx.Now(chasmComponent)
3880
secondNow := ctx.Now(chasmComponent)
3881
s.Equal(firstNow, secondNow)
3882
3883
observedTimes = append(observedTimes, firstNow)
3884
if len(observedTimes) == 1 {
3885
s.timeSource.Update(nextTaskTime)
3886
}
3887
return nil
3888
}).
3889
Times(2)
3890
3891
mutations, err := root.CloseTransaction()
3892
s.NoError(err)
3893
s.Empty(mutations.DeletedNodes)
3894
s.Equal([]time.Time{taskStartTime, nextTaskTime}, observedTimes)
3895
}
3896
3897
func (s *nodeSuite) TestEachPureTask() {
3898
now := s.timeSource.Now()
3899
3900
mustEncode := func(m proto.Message) *commonpb.DataBlob {
3901
taskBlob, err := encodeChasmBlob(m)
3902
s.NoError(err)
3903
return taskBlob
3904
}
3905
3906
// Set up a tree with expired and unexpired pure tasks.
3907
persistenceNodes := map[string]*persistencespb.ChasmNode{
3908
"": {
3909
Metadata: &persistencespb.ChasmNodeMetadata{
3910
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3911
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3912
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3913
TypeId: testComponentTypeID,
3914
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
3915
{
3916
// Expired
3917
TypeId: testPureTaskTypeID,
3918
ScheduledTime: timestamppb.New(now),
3919
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3920
VersionedTransitionOffset: 1,
3921
PhysicalTaskStatus: physicalTaskStatusCreated,
3922
Data: mustEncode(&commonpb.Payload{
3923
Data: []byte("some-random-data-root"),
3924
}),
3925
},
3926
},
3927
},
3928
},
3929
},
3930
},
3931
"SubComponent1": {
3932
Metadata: &persistencespb.ChasmNodeMetadata{
3933
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3934
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3935
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3936
TypeId: testSubComponent1TypeID,
3937
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
3938
{
3939
TypeId: testPureTaskTypeID,
3940
// Not expired yet.
3941
ScheduledTime: timestamppb.New(now.Add(time.Hour)),
3942
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3943
VersionedTransitionOffset: 2,
3944
PhysicalTaskStatus: physicalTaskStatusCreated,
3945
Data: mustEncode(&commonpb.Payload{
3946
Data: []byte("some-random-data-sc1"),
3947
}),
3948
},
3949
},
3950
},
3951
},
3952
},
3953
},
3954
"SubComponent1/SubComponent11": {
3955
Metadata: &persistencespb.ChasmNodeMetadata{
3956
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3957
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
3958
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
3959
TypeId: testSubComponent11TypeID,
3960
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
3961
{
3962
TypeId: testPureTaskTypeID,
3963
// Expired, and physical task not created
3964
ScheduledTime: timestamppb.New(now),
3965
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3966
VersionedTransitionOffset: 3,
3967
PhysicalTaskStatus: physicalTaskStatusNone,
3968
Data: mustEncode(&commonpb.Payload{
3969
Data: []byte("some-random-data-sc11-1"),
3970
}),
3971
},
3972
{
3973
TypeId: testPureTaskTypeID,
3974
// Expired, but when processing this task, delete the SubComponent11 itself.
3975
ScheduledTime: timestamppb.New(now),
3976
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3977
VersionedTransitionOffset: 4,
3978
PhysicalTaskStatus: physicalTaskStatusCreated,
3979
Data: mustEncode(&commonpb.Payload{
3980
Data: []byte("some-random-data-sc11-2"),
3981
}),
3982
},
3983
{
3984
TypeId: testPureTaskTypeID,
3985
// Expired, but should not be executed because previous task deletes SubComponent1
3986
// (this node's parent).
3987
ScheduledTime: timestamppb.New(now),
3988
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
3989
VersionedTransitionOffset: 5,
3990
PhysicalTaskStatus: physicalTaskStatusCreated,
3991
Data: mustEncode(&commonpb.Payload{
3992
Data: []byte("some-random-data-sc11-3"),
3993
}),
3994
},
3995
},
3996
},
3997
},
3998
},
3999
},
4000
"SubComponent2": {
4001
Metadata: &persistencespb.ChasmNodeMetadata{
4002
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
4003
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
4004
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
4005
TypeId: testSubComponent2TypeID,
4006
PureTasks: []*persistencespb.ChasmComponentAttributes_Task{
4007
{
4008
TypeId: testPureTaskTypeID,
4009
// Expired. However, this task won't be executed because the node is deleted
4010
// when processing the pure task from the root component.
4011
ScheduledTime: timestamppb.New(now),
4012
VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
4013
VersionedTransitionOffset: 6,
4014
PhysicalTaskStatus: physicalTaskStatusCreated,
4015
Data: mustEncode(&commonpb.Payload{
4016
Data: []byte("some-random-data-sc2"),
4017
}),
4018
},
4019
},
4020
},
4021
},
4022
},
4023
},
4024
}
4025
4026
root, err := s.newTestTree(persistenceNodes)
4027
s.NoError(err)
4028
s.NotNil(root)
4029
4030
processedTaskData := [][]byte{}
4031
err = root.EachPureTask(now.Add(time.Minute), func(handler NodePureTask, taskAttributes TaskAttributes, task any) (bool, error) {
4032
s.NotNil(handler)
4033
s.NotNil(taskAttributes)
4034
4035
testPureTask, ok := task.(*TestPureTask)
4036
s.True(ok)
4037
4038
processedTaskData = append(processedTaskData, testPureTask.Data)
4039
4040
// When processing root component task, delete SubComponent2 to verify its task is not executed.
4041
if slices.Equal(
4042
testPureTask.Data,
4043
[]byte("some-random-data-root"),
4044
) {
4045
mutableContext := NewMutableContext(context.Background(), root)
4046
rootComponent, err := root.Component(mutableContext, ComponentRef{})
4047
s.NoError(err)
4048
4049
rootComponent.(*TestComponent).SubComponent2 = NewEmptyField[*TestSubComponent2]()
4050
}
4051
4052
// When processing task for SubComponent11, delete its parent SubComponent1 so that the remaining task is not executed.
4053
if slices.Equal(
4054
testPureTask.Data,
4055
[]byte("some-random-data-sc11-2"),
4056
) {
4057
mutableContext := NewMutableContext(context.Background(), root)
4058
rootComponent, err := root.Component(mutableContext, ComponentRef{})
4059
s.NoError(err)
4060
4061
rootComponent.(*TestComponent).SubComponent1 = NewEmptyField[*TestSubComponent1]()
4062
}
4063
4064
return true, nil
4065
})
4066
s.NoError(err)
4067
s.Equal([][]byte{
4068
[]byte("some-random-data-root"),
4069
[]byte("some-random-data-sc11-1"),
4070
[]byte("some-random-data-sc11-2"),
4071
}, processedTaskData)
4072
s.Len(root.taskValueCache, 1) // only one task from root component
4073
}
4074
4075
func (s *nodeSuite) TestExecutePureTask() {
4076
persistenceNodes := map[string]*persistencespb.ChasmNode{
4077
"": {
4078
Metadata: &persistencespb.ChasmNodeMetadata{
4079
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
4080
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
4081
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
4082
TypeId: testComponentTypeID,
4083
},
4084
},
4085
},
4086
},
4087
}
4088
4089
taskAttributes := TaskAttributes{}
4090
pureTask := &TestPureTask{
4091
Data: []byte("some-random-data"),
4092
}
4093
4094
root, err := s.newTestTree(persistenceNodes)
4095
s.NoError(err)
4096
s.NotNil(root)
4097
ctx := context.Background()
4098
4099
expectExecute := func(result error) {
4100
s.testLibrary.mockPureTaskHandler.EXPECT().
4101
Execute(
4102
gomock.AssignableToTypeOf(&mutableCtx{}),
4103
gomock.AssignableToTypeOf(&TestComponent{}),
4104
gomock.Eq(taskAttributes),
4105
gomock.Eq(pureTask),
4106
).Return(result).Times(1)
4107
}
4108
4109
expectValidate := func(retValue bool, errValue error) {
4110
s.testLibrary.mockPureTaskHandler.EXPECT().
4111
Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: taskAttributes}), gomock.Eq(pureTask)).
4112
Return(retValue, errValue).
4113
Times(1)
4114
}
4115
4116
// Succeed task execution and validation (happy case).
4117
root.setValueState(valueStateSynced)
4118
expectExecute(nil)
4119
expectValidate(true, nil)
4120
executed, err := root.ExecutePureTask(ctx, taskAttributes, pureTask)
4121
s.NoError(err)
4122
s.True(executed)
4123
s.Equal(valueStateNeedSyncStructure, root.valueState)
4124
4125
expectedErr := errors.New("dummy")
4126
4127
// Succeed validation, fail execution.
4128
root.setValueState(valueStateSynced)
4129
expectExecute(expectedErr)
4130
expectValidate(true, nil)
4131
_, err = root.ExecutePureTask(ctx, taskAttributes, pureTask)
4132
s.ErrorIs(expectedErr, err)
4133
s.Equal(valueStateNeedSyncStructure, root.valueState)
4134
4135
// Fail task validation (no execution occurs).
4136
root.setValueState(valueStateSynced)
4137
expectValidate(false, nil)
4138
executed, err = root.ExecutePureTask(ctx, taskAttributes, pureTask)
4139
s.NoError(err)
4140
s.False(executed)
4141
s.Equal(valueStateSynced, root.valueState)
4142
s.True(root.subtreeIsDirty)
4143
4144
// Error during task validation (no execution occurs).
4145
root.setValueState(valueStateSynced)
4146
expectValidate(false, expectedErr)
4147
_, err = root.ExecutePureTask(ctx, taskAttributes, pureTask)
4148
s.ErrorIs(expectedErr, err)
4149
s.Equal(valueStateSynced, root.valueState) // task not executed, so node is clean
4150
}
4151
4152
func (s *nodeSuite) TestExecuteSideEffectTask() {
4153
persistenceNodes := map[string]*persistencespb.ChasmNode{
4154
"": {
4155
Metadata: &persistencespb.ChasmNodeMetadata{
4156
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
4157
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
4158
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
4159
TypeId: testComponentTypeID,
4160
},
4161
},
4162
},
4163
},
4164
"SubComponent1": {
4165
Metadata: &persistencespb.ChasmNodeMetadata{
4166
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
4167
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
4168
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
4169
TypeId: testSubComponent1TypeID,
4170
},
4171
},
4172
},
4173
},
4174
}
4175
4176
emptyTaskBlob := s.emptyDataBlob()
4177
taskInfo := &persistencespb.ChasmTaskInfo{
4178
ComponentInitialVersionedTransition: &persistencespb.VersionedTransition{
4179
TransitionCount: 1,
4180
},
4181
ComponentLastUpdateVersionedTransition: &persistencespb.VersionedTransition{
4182
TransitionCount: 1,
4183
},
4184
Path: []string{"SubComponent1"},
4185
TypeId: testSideEffectTaskTypeID,
4186
ArchetypeId: testComponentTypeID,
4187
Data: emptyTaskBlob,
4188
}
4189
workflowKey := definition.NewWorkflowKey(
4190
primitives.NewUUID().String(),
4191
primitives.NewUUID().String(),
4192
primitives.NewUUID().String(),
4193
)
4194
chasmTask := &tasks.ChasmTask{
4195
WorkflowKey: workflowKey,
4196
VisibilityTimestamp: s.timeSource.Now(),
4197
TaskID: 123,
4198
Category: tasks.CategoryOutbound,
4199
Destination: "destination",
4200
Info: taskInfo,
4201
}
4202
executionKey := ExecutionKey{
4203
NamespaceID: chasmTask.NamespaceID,
4204
BusinessID: chasmTask.WorkflowID,
4205
RunID: chasmTask.RunID,
4206
}
4207
4208
root, err := s.newTestTree(persistenceNodes)
4209
s.NoError(err)
4210
s.NotNil(root)
4211
4212
mockEngine := NewMockEngine(s.controller)
4213
ctx := NewEngineContext(context.Background(), mockEngine)
4214
4215
chasmContext := NewMutableContext(ctx, root)
4216
var backendValidtionFnCalled bool
4217
// This won't be called until access time.
4218
dummyValidationFn := func(_ NodeBackend, _ Context, _ Component) error {
4219
backendValidtionFnCalled = true
4220
return nil
4221
}
4222
expectValidate := func(valid bool, validationErr error) {
4223
backendValidtionFnCalled = false
4224
s.testLibrary.mockSideEffectTaskHandler.EXPECT().Validate(
4225
gomock.Any(),
4226
gomock.Any(),
4227
gomock.Any(),
4228
gomock.Any(),
4229
).Return(valid, validationErr).Times(1)
4230
}
4231
expectExecute := func(result error) {
4232
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
4233
Execute(
4234
gomock.Any(),
4235
gomock.Any(),
4236
gomock.Eq(TaskAttributes{
4237
ScheduledTime: chasmTask.GetVisibilityTime(),
4238
Destination: chasmTask.Destination,
4239
}),
4240
gomock.Any(),
4241
).DoAndReturn(
4242
func(_ context.Context, ref ComponentRef, _ TaskAttributes, _ *TestSideEffectTask) error {
4243
s.NotNil(ref.validationFn)
4244
s.Equal(taskInfo.GetArchetypeId(), uint32(ref.archetypeID))
4245
4246
// Accessing the Component should trigger the validationFn.
4247
component, err := root.Component(chasmContext, ref)
4248
if err != nil {
4249
return err
4250
}
4251
s.IsType(&TestSubComponent1{}, component)
4252
return result
4253
}).Times(1)
4254
}
4255
4256
// Succeed task execution.
4257
expectValidate(true, nil)
4258
expectExecute(nil)
4259
err = root.ExecuteSideEffectTask(ctx, executionKey, chasmTask, dummyValidationFn)
4260
s.NoError(err)
4261
s.True(backendValidtionFnCalled)
4262
s.True(chasmTask.DeserializedTask.IsValid())
4263
4264
// Invalid task.
4265
expectValidate(false, nil)
4266
expectExecute(nil)
4267
err = root.ExecuteSideEffectTask(ctx, executionKey, chasmTask, dummyValidationFn)
4268
s.Error(err)
4269
s.IsType(&serviceerror.NotFound{}, err)
4270
s.True(chasmTask.DeserializedTask.IsValid())
4271
4272
// Failed to validate task.
4273
validationErr := errors.New("validation error")
4274
expectValidate(false, validationErr)
4275
expectExecute(nil)
4276
err = root.ExecuteSideEffectTask(ctx, executionKey, chasmTask, dummyValidationFn)
4277
s.ErrorIs(validationErr, err)
4278
s.False(chasmTask.DeserializedTask.IsValid())
4279
4280
// Fail task execution.
4281
expectValidate(true, nil)
4282
executionErr := errors.New("execution error")
4283
expectExecute(executionErr)
4284
err = root.ExecuteSideEffectTask(ctx, executionKey, chasmTask, dummyValidationFn)
4285
s.ErrorIs(executionErr, err)
4286
s.True(backendValidtionFnCalled)
4287
s.False(chasmTask.DeserializedTask.IsValid())
4288
}
4289
4290
func (s *nodeSuite) TestExecuteSideEffectDiscardTask() {
4291
setup := func() (*Node, *tasks.ChasmTask, ExecutionKey, context.Context, Context) {
4292
persistenceNodes := map[string]*persistencespb.ChasmNode{
4293
"": {
4294
Metadata: &persistencespb.ChasmNodeMetadata{
4295
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
4296
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
4297
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
4298
TypeId: testComponentTypeID,
4299
},
4300
},
4301
},
4302
},
4303
"SubComponent1": {
4304
Metadata: &persistencespb.ChasmNodeMetadata{
4305
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
4306
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
4307
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
4308
TypeId: testSubComponent1TypeID,
4309
},
4310
},
4311
},
4312
},
4313
}
4314
4315
root, err := s.newTestTree(persistenceNodes)
4316
s.NoError(err)
4317
s.NotNil(root)
4318
4319
workflowKey := definition.NewWorkflowKey(
4320
primitives.NewUUID().String(),
4321
primitives.NewUUID().String(),
4322
primitives.NewUUID().String(),
4323
)
4324
emptyTaskBlob := s.emptyDataBlob()
4325
chasmTask := &tasks.ChasmTask{
4326
WorkflowKey: workflowKey,
4327
VisibilityTimestamp: s.timeSource.Now(),
4328
TaskID: 123,
4329
Category: tasks.CategoryOutbound,
4330
Destination: "destination",
4331
Info: &persistencespb.ChasmTaskInfo{
4332
ComponentInitialVersionedTransition: &persistencespb.VersionedTransition{
4333
TransitionCount: 1,
4334
},
4335
ComponentLastUpdateVersionedTransition: &persistencespb.VersionedTransition{
4336
TransitionCount: 1,
4337
},
4338
Path: []string{"SubComponent1"},
4339
TypeId: testDiscardableSideEffectTaskTypeID,
4340
ArchetypeId: testComponentTypeID,
4341
Data: emptyTaskBlob,
4342
},
4343
}
4344
executionKey := ExecutionKey{
4345
NamespaceID: chasmTask.NamespaceID,
4346
BusinessID: chasmTask.WorkflowID,
4347
RunID: chasmTask.RunID,
4348
}
4349
4350
mockEngine := NewMockEngine(s.controller)
4351
ctx := NewEngineContext(context.Background(), mockEngine)
4352
chasmContext := NewMutableContext(ctx, root)
4353
4354
return root, chasmTask, executionKey, ctx, chasmContext
4355
}
4356
4357
s.Run("Success", func() {
4358
root, chasmTask, executionKey, ctx, chasmContext := setup()
4359
4360
var validationFnCalled bool
4361
dummyValidationFn := func(_ NodeBackend, _ Context, _ Component) error {
4362
validationFnCalled = true
4363
return nil
4364
}
4365
4366
s.testLibrary.mockDiscardableSideEffectHandler.EXPECT().Validate(
4367
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
4368
).Return(true, nil).Times(1)
4369
s.testLibrary.mockDiscardableSideEffectHandler.EXPECT().Discard(
4370
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
4371
).DoAndReturn(func(
4372
_ context.Context, ref ComponentRef, _ TaskAttributes, _ *TestDiscardableSideEffectTask,
4373
) error {
4374
s.NotNil(ref.validationFn)
4375
s.Equal(chasmTask.Info.GetArchetypeId(), uint32(ref.archetypeID))
4376
component, err := root.Component(chasmContext, ref)
4377
if err != nil {
4378
return err
4379
}
4380
s.IsType(&TestSubComponent1{}, component)
4381
return nil
4382
}).Times(1)
4383
4384
err := root.ExecuteSideEffectDiscardTask(ctx, executionKey, chasmTask, dummyValidationFn)
4385
s.NoError(err)
4386
s.True(validationFnCalled)
4387
s.True(chasmTask.DeserializedTask.IsValid())
4388
})
4389
4390
s.Run("InvalidTask", func() {
4391
root, chasmTask, executionKey, ctx, chasmContext := setup()
4392
4393
s.testLibrary.mockDiscardableSideEffectHandler.EXPECT().Validate(
4394
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
4395
).Return(false, nil).Times(1)
4396
s.testLibrary.mockDiscardableSideEffectHandler.EXPECT().Discard(
4397
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
4398
).DoAndReturn(func(
4399
_ context.Context, ref ComponentRef, _ TaskAttributes, _ *TestDiscardableSideEffectTask,
4400
) error {
4401
_, err := root.Component(chasmContext, ref)
4402
return err
4403
}).Times(1)
4404
4405
err := root.ExecuteSideEffectDiscardTask(ctx, executionKey, chasmTask, func(_ NodeBackend, _ Context, _ Component) error { return nil })
4406
s.ErrorAs(err, new(*serviceerror.NotFound))
4407
})
4408
4409
s.Run("ValidationError", func() {
4410
root, chasmTask, executionKey, ctx, chasmContext := setup()
4411
4412
validationErr := errors.New("validation error")
4413
s.testLibrary.mockDiscardableSideEffectHandler.EXPECT().Validate(
4414
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
4415
).Return(false, validationErr).Times(1)
4416
s.testLibrary.mockDiscardableSideEffectHandler.EXPECT().Discard(
4417
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
4418
).DoAndReturn(func(
4419
_ context.Context, ref ComponentRef, _ TaskAttributes, _ *TestDiscardableSideEffectTask,
4420
) error {
4421
_, err := root.Component(chasmContext, ref)
4422
return err
4423
}).Times(1)
4424
4425
err := root.ExecuteSideEffectDiscardTask(
4426
ctx, executionKey, chasmTask, func(_ NodeBackend, _ Context, _ Component) error { return nil })
4427
s.ErrorIs(err, validationErr)
4428
})
4429
4430
s.Run("DiscardHandlerError", func() {
4431
root, chasmTask, executionKey, ctx, chasmContext := setup()
4432
4433
var validationFnCalled bool
4434
dummyValidationFn := func(_ NodeBackend, _ Context, _ Component) error {
4435
validationFnCalled = true
4436
return nil
4437
}
4438
4439
s.testLibrary.mockDiscardableSideEffectHandler.EXPECT().Validate(
4440
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
4441
).Return(true, nil).Times(1)
4442
discardErr := errors.New("discard error")
4443
s.testLibrary.mockDiscardableSideEffectHandler.EXPECT().Discard(
4444
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
4445
).DoAndReturn(func(
4446
_ context.Context, ref ComponentRef, _ TaskAttributes, _ *TestDiscardableSideEffectTask,
4447
) error {
4448
s.NotNil(ref.validationFn)
4449
if _, err := root.Component(chasmContext, ref); err != nil {
4450
return err
4451
}
4452
return discardErr
4453
}).Times(1)
4454
4455
err := root.ExecuteSideEffectDiscardTask(ctx, executionKey, chasmTask, dummyValidationFn)
4456
s.ErrorIs(err, discardErr)
4457
s.True(validationFnCalled)
4458
})
4459
}
4460
4461
func (s *nodeSuite) TestValidateSideEffectTask() {
4462
emptyTaskBlob := s.emptyDataBlob()
4463
taskInfo := &persistencespb.ChasmTaskInfo{
4464
ComponentInitialVersionedTransition: &persistencespb.VersionedTransition{
4465
TransitionCount: 1,
4466
NamespaceFailoverVersion: 1,
4467
},
4468
ComponentLastUpdateVersionedTransition: &persistencespb.VersionedTransition{
4469
TransitionCount: 1,
4470
NamespaceFailoverVersion: 1,
4471
},
4472
Path: rootPath,
4473
TypeId: testSideEffectTaskTypeID,
4474
Data: emptyTaskBlob,
4475
}
4476
workflowKey := definition.NewWorkflowKey(
4477
primitives.NewUUID().String(),
4478
primitives.NewUUID().String(),
4479
primitives.NewUUID().String(),
4480
)
4481
chasmTask := &tasks.ChasmTask{
4482
WorkflowKey: workflowKey,
4483
VisibilityTimestamp: s.timeSource.Now(),
4484
TaskID: 123,
4485
Category: tasks.CategoryTransfer,
4486
Info: taskInfo,
4487
}
4488
4489
root := s.testComponentTree()
4490
4491
mockEngine := NewMockEngine(s.controller)
4492
ctx := NewEngineContext(context.Background(), mockEngine)
4493
4494
expectValidate := func(componentType any, retValue bool, errValue error) {
4495
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
4496
Validate(
4497
gomock.AssignableToTypeOf((*immutableCtx)(nil)),
4498
gomock.AssignableToTypeOf(componentType),
4499
gomock.Eq(TaskInvocation{
4500
TaskAttributes: TaskAttributes{
4501
ScheduledTime: chasmTask.GetVisibilityTime(),
4502
Destination: chasmTask.Destination,
4503
},
4504
}),
4505
gomock.AssignableToTypeOf(&TestSideEffectTask{}),
4506
).Return(retValue, errValue).Times(1)
4507
}
4508
4509
// Succeed validation as valid.
4510
expectValidate((*TestComponent)(nil), true, nil)
4511
isTaskInTree, isValidByComponent, err := root.ValidateSideEffectTask(ctx, chasmTask)
4512
s.True(isTaskInTree)
4513
s.True(isValidByComponent)
4514
s.NoError(err)
4515
s.True(chasmTask.DeserializedTask.IsValid())
4516
4517
// The physical task's attempt is threaded into the validator's TaskInvocation.
4518
chasmTask.Attempt = 7
4519
s.testLibrary.mockSideEffectTaskHandler.EXPECT().
4520
Validate(
4521
gomock.AssignableToTypeOf((*immutableCtx)(nil)),
4522
gomock.AssignableToTypeOf((*TestComponent)(nil)),
4523
gomock.Any(),
4524
gomock.AssignableToTypeOf(&TestSideEffectTask{}),
4525
).DoAndReturn(func(_ Context, _ any, inv TaskInvocation, _ *TestSideEffectTask) (bool, error) {
4526
s.Equal(7, inv.Attempt)
4527
return true, nil
4528
}).Times(1)
4529
isTaskInTree, isValidByComponent, err = root.ValidateSideEffectTask(ctx, chasmTask)
4530
s.True(isTaskInTree)
4531
s.True(isValidByComponent)
4532
s.NoError(err)
4533
chasmTask.Attempt = 0
4534
4535
// Task is in tree but component says invalid.
4536
expectValidate((*TestComponent)(nil), false, nil)
4537
isTaskInTree, isValidByComponent, err = root.ValidateSideEffectTask(ctx, chasmTask)
4538
s.True(isTaskInTree)
4539
s.False(isValidByComponent)
4540
s.NoError(err)
4541
s.True(chasmTask.DeserializedTask.IsValid())
4542
4543
// Component validator returns an error — task was found in the tree, but validation failed.
4544
expectedErr := errors.New("validation failed")
4545
expectValidate((*TestComponent)(nil), false, expectedErr)
4546
isTaskInTree, isValidByComponent, err = root.ValidateSideEffectTask(ctx, chasmTask)
4547
s.True(isTaskInTree)
4548
s.False(isValidByComponent)
4549
s.ErrorIs(expectedErr, err)
4550
s.False(chasmTask.DeserializedTask.IsValid())
4551
4552
// Succeed validation as valid for a sub component.
4553
childTaskInfo := taskInfo
4554
childTaskInfo.Path = []string{"SubComponent1"}
4555
childWorkflowKey := definition.NewWorkflowKey(
4556
primitives.NewUUID().String(),
4557
primitives.NewUUID().String(),
4558
primitives.NewUUID().String(),
4559
)
4560
childChasmTask := &tasks.ChasmTask{
4561
WorkflowKey: childWorkflowKey,
4562
VisibilityTimestamp: s.timeSource.Now(),
4563
TaskID: 124,
4564
Category: tasks.CategoryTransfer,
4565
Info: childTaskInfo,
4566
}
4567
expectValidate((*TestSubComponent1)(nil), true, nil)
4568
isTaskInTree, isValidByComponent, err = root.ValidateSideEffectTask(ctx, childChasmTask)
4569
s.True(isTaskInTree)
4570
s.True(isValidByComponent)
4571
s.NoError(err)
4572
s.True(childChasmTask.DeserializedTask.IsValid())
4573
4574
// Component access check fails (parent closed) — task is structurally in the tree but
4575
// isValidByComponent=false because the access rule rejects it.
4576
mutableCtx := NewMutableContext(ctx, root)
4577
rootComponent, err := root.ComponentByPath(mutableCtx, rootPath)
4578
s.NoError(err)
4579
rootComponent.(*TestComponent).Complete(mutableCtx)
4580
// Note there's also no mock for the task validator here; the access rule is checked first.
4581
isTaskInTree, isValidByComponent, err = root.ValidateSideEffectTask(ctx, childChasmTask)
4582
s.True(isTaskInTree)
4583
s.False(isValidByComponent)
4584
s.NoError(err)
4585
s.True(childChasmTask.DeserializedTask.IsValid())
4586
}
4587
4588
func (s *nodeSuite) TestAndAllChildren_PathIndependence() {
4589
// Build a tree deep enough to trigger Go's slice capacity doubling.
4590
// append grows cap: 0→1→2→4. At depth 3, the path slice has len=3, cap=4,
4591
// so a 4th append reuses the backing array. If node P at depth 3 has siblings
4592
// S1 and S2 at depth 4, the second sibling's append overwrites S1's path.
4593
//
4594
// Tree: root → A → B → C → {S1, S2}
4595
root := &Node{
4596
nodeName: "",
4597
children: map[string]*Node{
4598
"A": {nodeName: "A", children: map[string]*Node{
4599
"B": {nodeName: "B", children: map[string]*Node{
4600
"C": {nodeName: "C", children: map[string]*Node{
4601
"S1": {nodeName: "S1", children: map[string]*Node{}},
4602
"S2": {nodeName: "S2", children: map[string]*Node{}},
4603
}},
4604
}},
4605
}},
4606
},
4607
}
4608
4609
// Store raw path slices (not copies!) so we can detect mutation.
4610
collected := make(map[string][]string)
4611
for path, node := range root.andAllChildren() {
4612
collected[node.nodeName] = path
4613
}
4614
4615
// Verify S1/S2 do not have a corrupted path
4616
// because append reused the backing array at depth 3→4.
4617
s.Equal([]string{"A", "B", "C", "S1"}, collected["S1"])
4618
s.Equal([]string{"A", "B", "C", "S2"}, collected["S2"])
4619
}
4620
4621
func (s *nodeSuite) newTestTree(
4622
serializedNodes map[string]*persistencespb.ChasmNode,
4623
) (*Node, error) {
4624
if len(serializedNodes) == 0 {
4625
return NewEmptyTree(s.registry, s.timeSource, s.nodeBackend, s.nodePathEncoder, s.logger, s.metricsHandler), nil
4626
}
4627
return NewTreeFromDB(serializedNodes, s.registry, s.timeSource, s.nodeBackend, s.nodePathEncoder, s.logger, s.metricsHandler)
4628
}
4629
4630
func (s *nodeSuite) emptyDataBlob() *commonpb.DataBlob {
4631
blob, err := encodeChasmBlob(nil)
4632
s.NoError(err)
4633
return blob
4634
}
4635
4636
type protoMatcher struct {
4637
x proto.Message
4638
}
4639
4640
func (e protoMatcher) Matches(x any) bool {
4641
if xx, ok := x.(proto.Message); ok {
4642
return proto.Equal(e.x, xx)
4643
}
4644
return false
4645
}
4646
4647
func (e protoMatcher) String() string {
4648
return fmt.Sprintf("is proto equal to %s (%T)", e.x, e.x)
4649
}
4650
4651
func protoEq(x proto.Message) gomock.Matcher {
4652
return protoMatcher{x: x}
4653
}
4654
4655
// TestCloseTransaction_AppliesPendingComponentMetadata verifies that
4656
// SetRequestLinks/SetUserMetadata writes are written onto the root component's
4657
// ChasmComponentAttributes during CloseTransaction, that the touched node is
4658
// added to NodesMutation.UpdatedNodes, and that its LastUpdateVersionedTransition
4659
// is bumped.
4660
func (s *nodeSuite) TestCloseTransaction_AppliesPendingComponentMetadata() {
4661
const requestID = "req-1"
4662
link := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{
4663
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"},
4664
}}
4665
md := &sdkpb.UserMetadata{Summary: &commonpb.Payload{Data: []byte("summary")}}
4666
4667
root := s.testComponentTree() // sets HandleNextTransitionCount = 1, HandleGetCurrentVersion = 1
4668
4669
// Initial create transaction must close cleanly before we exercise the metadata path.
4670
_, err := root.CloseTransaction()
4671
s.NoError(err)
4672
4673
// Bump the transition count so we can verify LastUpdateVersionedTransition was updated.
4674
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
4675
4676
ctx := NewMutableContext(context.Background(), root)
4677
c, err := root.Component(ctx, ComponentRef{})
4678
s.NoError(err)
4679
4680
s.NoError(ctx.SetRequestLinks(c, requestID, []*commonpb.Link{link}))
4681
s.NoError(ctx.SetUserMetadata(c, md))
4682
4683
mutation, err := root.CloseTransaction()
4684
s.NoError(err)
4685
4686
rootSerialized, ok := mutation.UpdatedNodes[""]
4687
s.True(ok, "root node must appear in UpdatedNodes after staging metadata")
4688
attrs := rootSerialized.GetMetadata().GetComponentAttributes()
4689
s.NotNil(attrs)
4690
s.Equal([]*commonpb.Link{link}, attrs.GetRequests()[requestID].GetLinks())
4691
s.Equal(md.GetSummary().GetData(), attrs.GetUserMetadata().GetSummary().GetData())
4692
s.Equal(int64(2), rootSerialized.GetMetadata().GetLastUpdateVersionedTransition().GetTransitionCount())
4693
4694
// Pending maps must be cleared after CloseTransaction so a subsequent transaction
4695
// does not re-apply the same writes.
4696
s.Empty(root.pendingRequestLinks)
4697
s.Empty(root.pendingUserMetadata)
4698
}
4699
4700
// TestSetComponentMetadata_MarksTreeDirty verifies that staging a
4701
// SetRequestLinks or SetUserMetadata write flips IsDirty()/IsStateDirty()
4702
// before CloseTransaction runs.
4703
func (s *nodeSuite) TestSetComponentMetadata_MarksTreeDirty() {
4704
root := s.testComponentTree()
4705
_, err := root.CloseTransaction()
4706
s.NoError(err)
4707
4708
s.False(root.IsDirty(), "tree must be clean after the initial close")
4709
s.False(root.IsStateDirty())
4710
4711
ctx := NewMutableContext(context.Background(), root)
4712
c, err := root.Component(ctx, ComponentRef{})
4713
s.NoError(err)
4714
4715
s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{{
4716
Variant: &commonpb.Link_WorkflowEvent_{
4717
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"},
4718
},
4719
}}))
4720
s.True(root.IsStateDirty(), "staging SetRequestLinks must mark the tree dirty")
4721
s.True(root.IsDirty())
4722
4723
_, err = root.CloseTransaction()
4724
s.NoError(err)
4725
s.False(root.IsStateDirty(), "CloseTransaction must clear the dirty flag")
4726
4727
ctx = NewMutableContext(context.Background(), root)
4728
c, err = root.Component(ctx, ComponentRef{})
4729
s.NoError(err)
4730
s.NoError(ctx.SetUserMetadata(c, &sdkpb.UserMetadata{
4731
Summary: &commonpb.Payload{Data: []byte("summary")},
4732
}))
4733
s.True(root.IsStateDirty(), "staging SetUserMetadata must mark the tree dirty")
4734
s.True(root.IsDirty())
4735
}
4736
4737
// TestCloseTransaction_DropsOrphanedComponentMetadata verifies that pending
4738
// SetRequestLinks/SetUserMetadata writes against a component value that is
4739
// not registered in the tree are silently dropped during CloseTransaction
4740
// (rather than panicking or surfacing an error), and that the pending maps
4741
// are cleared afterwards.
4742
func (s *nodeSuite) TestCloseTransaction_DropsOrphanedComponentMetadata() {
4743
root := s.testComponentTree()
4744
_, err := root.CloseTransaction()
4745
s.NoError(err)
4746
4747
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
4748
ctx := NewMutableContext(context.Background(), root)
4749
4750
// Stage writes against a component value that was never set on the tree.
4751
orphan := &TestComponent{}
4752
s.NoError(ctx.SetRequestLinks(orphan, "req-id", []*commonpb.Link{{
4753
Variant: &commonpb.Link_WorkflowEvent_{
4754
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"},
4755
},
4756
}}))
4757
s.NoError(ctx.SetUserMetadata(orphan, &sdkpb.UserMetadata{
4758
Summary: &commonpb.Payload{Data: []byte("orphan")},
4759
}))
4760
4761
mutation, err := root.CloseTransaction()
4762
s.NoError(err)
4763
s.NotContains(mutation.UpdatedNodes, "", "root must not be updated by orphaned writes")
4764
s.Empty(root.pendingRequestLinks)
4765
s.Empty(root.pendingUserMetadata)
4766
}
4767
4768
// TestSetComponentRequestLinks_RejectsEmptyRequestID verifies the framework
4769
// hard-rejects empty requestIDs so two callers cannot silently collide on the
4770
// empty-string key.
4771
func (s *nodeSuite) TestSetComponentRequestLinks_RejectsEmptyRequestID() {
4772
root := s.testComponentTree()
4773
_, err := root.CloseTransaction()
4774
s.NoError(err)
4775
4776
ctx := NewMutableContext(context.Background(), root)
4777
c, err := root.Component(ctx, ComponentRef{})
4778
s.NoError(err)
4779
4780
err = ctx.SetRequestLinks(c, "", []*commonpb.Link{{
4781
Variant: &commonpb.Link_WorkflowEvent_{
4782
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"},
4783
},
4784
}})
4785
s.Error(err)
4786
s.ErrorAs(err, new(*serviceerror.InvalidArgument))
4787
4788
_, err = ctx.RequestLinks(c, "")
4789
s.Error(err)
4790
s.ErrorAs(err, new(*serviceerror.InvalidArgument))
4791
}
4792
4793
// TestSetRequestLinks_MultipleRequestsCoexist verifies that two distinct
4794
// request IDs on the same component land as separate entries in
4795
// ChasmComponentAttributes.requests.
4796
func (s *nodeSuite) TestSetRequestLinks_MultipleRequestsCoexist() {
4797
root := s.testComponentTree()
4798
_, err := root.CloseTransaction()
4799
s.NoError(err)
4800
4801
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
4802
ctx := NewMutableContext(context.Background(), root)
4803
c, err := root.Component(ctx, ComponentRef{})
4804
s.NoError(err)
4805
4806
linkA := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{
4807
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "a", RunId: "run"},
4808
}}
4809
linkB := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{
4810
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "b", RunId: "run"},
4811
}}
4812
s.NoError(ctx.SetRequestLinks(c, "req-a", []*commonpb.Link{linkA}))
4813
s.NoError(ctx.SetRequestLinks(c, "req-b", []*commonpb.Link{linkB}))
4814
4815
mutation, err := root.CloseTransaction()
4816
s.NoError(err)
4817
4818
attrs := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
4819
s.Len(attrs.GetRequests(), 2)
4820
s.Equal([]*commonpb.Link{linkA}, attrs.GetRequests()["req-a"].GetLinks())
4821
s.Equal([]*commonpb.Link{linkB}, attrs.GetRequests()["req-b"].GetLinks())
4822
}
4823
4824
// TestSetRequestLinks_ReplacesEntryForSameRequestID verifies that two
4825
// SetRequestLinks calls with the same requestID — within or across
4826
// transactions — leave only the second value in attrs.Requests.
4827
func (s *nodeSuite) TestSetRequestLinks_ReplacesEntryForSameRequestID() {
4828
root := s.testComponentTree()
4829
_, err := root.CloseTransaction()
4830
s.NoError(err)
4831
4832
linkA := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{
4833
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "a", RunId: "run"},
4834
}}
4835
linkB := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{
4836
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "b", RunId: "run"},
4837
}}
4838
4839
// Within a single transaction: second SetRequestLinks for the same requestID
4840
// must overwrite the first.
4841
nextTC := int64(2)
4842
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC }
4843
ctx := NewMutableContext(context.Background(), root)
4844
c, err := root.Component(ctx, ComponentRef{})
4845
s.NoError(err)
4846
s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{linkA}))
4847
s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{linkB}))
4848
mutation, err := root.CloseTransaction()
4849
s.NoError(err)
4850
s.Equal([]*commonpb.Link{linkB},
4851
mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes().GetRequests()["req"].GetLinks(),
4852
"second SetRequestLinks within a transaction must overwrite the first",
4853
)
4854
4855
// Across transactions: a later write for the same requestID must replace the
4856
// previously persisted entry.
4857
nextTC = 3
4858
ctx = NewMutableContext(context.Background(), root)
4859
c, err = root.Component(ctx, ComponentRef{})
4860
s.NoError(err)
4861
s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{linkA}))
4862
mutation, err = root.CloseTransaction()
4863
s.NoError(err)
4864
s.Equal([]*commonpb.Link{linkA},
4865
mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes().GetRequests()["req"].GetLinks(),
4866
"a follow-up transaction's SetRequestLinks must replace the persisted entry",
4867
)
4868
}
4869
4870
// TestSetRequestLinks_RemovesEntryWhenEmptyLinks verifies that passing nil/empty
4871
// links for a previously-stored requestID removes that entry from
4872
// ChasmComponentAttributes.requests.
4873
func (s *nodeSuite) TestSetRequestLinks_RemovesEntryWhenEmptyLinks() {
4874
root := s.testComponentTree()
4875
_, err := root.CloseTransaction()
4876
s.NoError(err)
4877
4878
link := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{
4879
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"},
4880
}}
4881
4882
// First persist a link under "req".
4883
nextTC := int64(2)
4884
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC }
4885
ctx := NewMutableContext(context.Background(), root)
4886
c, err := root.Component(ctx, ComponentRef{})
4887
s.NoError(err)
4888
s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{link}))
4889
mutation, err := root.CloseTransaction()
4890
s.NoError(err)
4891
s.Contains(mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes().GetRequests(), "req")
4892
4893
// Then clear it with an empty links slice.
4894
nextTC = 3
4895
ctx = NewMutableContext(context.Background(), root)
4896
c, err = root.Component(ctx, ComponentRef{})
4897
s.NoError(err)
4898
s.NoError(ctx.SetRequestLinks(c, "req", nil))
4899
mutation, err = root.CloseTransaction()
4900
s.NoError(err)
4901
attrs := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
4902
s.NotContains(attrs.GetRequests(), "req", "empty links must remove the entry for requestID")
4903
}
4904
4905
// TestRequestLinks_PrefersPendingOverPersisted verifies that an in-transaction
4906
// SetRequestLinks shadow-reads via RequestLinks / Links return the staged
4907
// (pending) value rather than the previously-persisted entry, so callers
4908
// reading-then-writing within a single transaction never observe stale state.
4909
func (s *nodeSuite) TestRequestLinks_PrefersPendingOverPersisted() {
4910
root := s.testComponentTree()
4911
_, err := root.CloseTransaction()
4912
s.NoError(err)
4913
4914
oldLink := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{
4915
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "old", RunId: "run"},
4916
}}
4917
newLink := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{
4918
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "new", RunId: "run"},
4919
}}
4920
4921
// Persist [oldLink] under "req".
4922
nextTC := int64(2)
4923
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC }
4924
ctx := NewMutableContext(context.Background(), root)
4925
c, err := root.Component(ctx, ComponentRef{})
4926
s.NoError(err)
4927
s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{oldLink}))
4928
_, err = root.CloseTransaction()
4929
s.NoError(err)
4930
4931
// Open a new transaction, stage a replace with [newLink] under the same
4932
// requestID, then read via both APIs before close.
4933
nextTC = 3
4934
ctx = NewMutableContext(context.Background(), root)
4935
c, err = root.Component(ctx, ComponentRef{})
4936
s.NoError(err)
4937
s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{newLink}))
4938
4939
got, err := ctx.RequestLinks(c, "req")
4940
s.NoError(err)
4941
s.Equal([]*commonpb.Link{newLink}, got, "RequestLinks must prefer pending over persisted for the same requestID")
4942
s.Equal([]*commonpb.Link{newLink}, ctx.Links(c),
4943
"Links must prefer pending and not return old+new duplicates for the same requestID")
4944
}
4945
4946
// TestCloseTransaction_PersistsAcrossTransactions verifies the realistic
4947
// production flow: write metadata in transaction A, commit, open transaction
4948
// B, read it back through the framework APIs.
4949
func (s *nodeSuite) TestCloseTransaction_PersistsAcrossTransactions() {
4950
root := s.testComponentTree()
4951
_, err := root.CloseTransaction()
4952
s.NoError(err)
4953
4954
link := &commonpb.Link{Variant: &commonpb.Link_WorkflowEvent_{
4955
WorkflowEvent: &commonpb.Link_WorkflowEvent{Namespace: "ns", WorkflowId: "wf", RunId: "run"},
4956
}}
4957
md := &sdkpb.UserMetadata{Summary: &commonpb.Payload{Data: []byte("summary")}}
4958
4959
nextTC := int64(2)
4960
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC }
4961
ctx := NewMutableContext(context.Background(), root)
4962
c, err := root.Component(ctx, ComponentRef{})
4963
s.NoError(err)
4964
s.NoError(ctx.SetRequestLinks(c, "req", []*commonpb.Link{link}))
4965
s.NoError(ctx.SetUserMetadata(c, md))
4966
_, err = root.CloseTransaction()
4967
s.NoError(err)
4968
4969
// New transaction: framework getters must surface the persisted attrs.
4970
nextTC = 3
4971
ctx2 := NewMutableContext(context.Background(), root)
4972
c2, err := root.Component(ctx2, ComponentRef{})
4973
s.NoError(err)
4974
4975
got, err := ctx2.RequestLinks(c2, "req")
4976
s.NoError(err)
4977
s.Equal([]*commonpb.Link{link}, got)
4978
s.Equal([]*commonpb.Link{link}, ctx2.Links(c2))
4979
s.ProtoEqual(md, ctx2.UserMetadata(c2))
4980
}
4981
4982
// TestSetUserMetadata_NilClearsPersistedValue verifies that SetUserMetadata
4983
// called with nil clears any previously-persisted user metadata on the
4984
// component (rather than being treated as a no-op).
4985
func (s *nodeSuite) TestSetUserMetadata_NilClearsPersistedValue() {
4986
root := s.testComponentTree()
4987
_, err := root.CloseTransaction()
4988
s.NoError(err)
4989
4990
// Persist user metadata.
4991
nextTC := int64(2)
4992
s.nodeBackend.HandleNextTransitionCount = func() int64 { return nextTC }
4993
ctx := NewMutableContext(context.Background(), root)
4994
c, err := root.Component(ctx, ComponentRef{})
4995
s.NoError(err)
4996
s.NoError(ctx.SetUserMetadata(c, &sdkpb.UserMetadata{
4997
Summary: &commonpb.Payload{Data: []byte("first")},
4998
}))
4999
_, err = root.CloseTransaction()
5000
s.NoError(err)
5001
5002
// Clear with nil.
5003
nextTC = 3
5004
ctx = NewMutableContext(context.Background(), root)
5005
c, err = root.Component(ctx, ComponentRef{})
5006
s.NoError(err)
5007
s.NoError(ctx.SetUserMetadata(c, nil))
5008
mutation, err := root.CloseTransaction()
5009
s.NoError(err)
5010
s.Nil(mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes().GetUserMetadata())
5011
}
5012
5013
func (s *nodeSuite) TestCloseTransaction_SingletonTask_Replace_SideEffect() {
5014
persistenceNodes := map[string]*persistencespb.ChasmNode{
5015
"": {
5016
Metadata: &persistencespb.ChasmNodeMetadata{
5017
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5018
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5019
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
5020
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
5021
TypeId: testComponentTypeID,
5022
},
5023
},
5024
},
5025
},
5026
}
5027
5028
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
5029
root, err := s.newTestTree(persistenceNodes)
5030
s.NoError(err)
5031
5032
// First transaction: add an initial singleton side-effect task.
5033
mutableContext := NewMutableContext(context.Background(), root)
5034
c, err := root.Component(mutableContext, ComponentRef{})
5035
s.NoError(err)
5036
testComponent := c.(*TestComponent)
5037
5038
s.testLibrary.mockSingletonReplaceSideEffectTaskHandler.EXPECT().
5039
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
5040
mutableContext.AddTask(testComponent, TaskAttributes{}, &TestSingletonReplaceSideEffectTask{Data: []byte("first")})
5041
5042
mutation, err := root.CloseTransaction()
5043
s.NoError(err)
5044
rootAttr := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5045
s.Len(rootAttr.SideEffectTasks, 1)
5046
s.Equal(testSingletonReplaceSideEffectTaskTypeID, rootAttr.SideEffectTasks[0].TypeId)
5047
5048
// Second transaction: add a second singleton task — it should replace the first.
5049
// closeTransactionCleanupInvalidTasks re-validates the existing task (returns true to keep it),
5050
// then closeTransactionHandleNewTasks validates the new task.
5051
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 3 }
5052
mutableContext = NewMutableContext(context.Background(), root)
5053
c, err = root.Component(mutableContext, ComponentRef{})
5054
s.NoError(err)
5055
testComponent = c.(*TestComponent)
5056
5057
s.testLibrary.mockSingletonReplaceSideEffectTaskHandler.EXPECT().
5058
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(2)
5059
mutableContext.AddTask(testComponent, TaskAttributes{}, &TestSingletonReplaceSideEffectTask{Data: []byte("second")})
5060
5061
mutation, err = root.CloseTransaction()
5062
s.NoError(err)
5063
rootAttr = mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5064
s.Len(rootAttr.SideEffectTasks, 1, "replace mode must keep exactly one task")
5065
s.Equal(testSingletonReplaceSideEffectTaskTypeID, rootAttr.SideEffectTasks[0].TypeId)
5066
}
5067
5068
func (s *nodeSuite) TestCloseTransaction_SingletonTask_Ignore_SideEffect() {
5069
persistenceNodes := map[string]*persistencespb.ChasmNode{
5070
"": {
5071
Metadata: &persistencespb.ChasmNodeMetadata{
5072
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5073
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5074
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
5075
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
5076
TypeId: testComponentTypeID,
5077
},
5078
},
5079
},
5080
},
5081
}
5082
5083
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
5084
root, err := s.newTestTree(persistenceNodes)
5085
s.NoError(err)
5086
5087
// First transaction: add the initial singleton side-effect task.
5088
mutableContext := NewMutableContext(context.Background(), root)
5089
c, err := root.Component(mutableContext, ComponentRef{})
5090
s.NoError(err)
5091
testComponent := c.(*TestComponent)
5092
5093
s.testLibrary.mockSingletonIgnoreSideEffectTaskHandler.EXPECT().
5094
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
5095
mutableContext.AddTask(testComponent, TaskAttributes{}, &TestSingletonIgnoreSideEffectTask{Data: []byte("first")})
5096
5097
mutation, err := root.CloseTransaction()
5098
s.NoError(err)
5099
rootAttr := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5100
s.Len(rootAttr.SideEffectTasks, 1)
5101
firstTask := rootAttr.SideEffectTasks[0]
5102
5103
// Second transaction: add a second singleton task — it should be discarded.
5104
// closeTransactionCleanupInvalidTasks re-validates the existing task (1 call),
5105
// then closeTransactionHandleNewTasks validates the new task (1 call).
5106
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 3 }
5107
mutableContext = NewMutableContext(context.Background(), root)
5108
c, err = root.Component(mutableContext, ComponentRef{})
5109
s.NoError(err)
5110
testComponent = c.(*TestComponent)
5111
5112
s.testLibrary.mockSingletonIgnoreSideEffectTaskHandler.EXPECT().
5113
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(2)
5114
mutableContext.AddTask(testComponent, TaskAttributes{}, &TestSingletonIgnoreSideEffectTask{Data: []byte("second")})
5115
5116
mutation, err = root.CloseTransaction()
5117
s.NoError(err)
5118
rootAttr = mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5119
s.Len(rootAttr.SideEffectTasks, 1, "ignore mode must keep exactly one task")
5120
s.Equal(firstTask.VersionedTransition, rootAttr.SideEffectTasks[0].VersionedTransition,
5121
"ignore mode must keep the original task, not replace it")
5122
}
5123
5124
func (s *nodeSuite) TestCloseTransaction_SingletonTask_Replace_Pure() {
5125
persistenceNodes := map[string]*persistencespb.ChasmNode{
5126
"": {
5127
Metadata: &persistencespb.ChasmNodeMetadata{
5128
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5129
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5130
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
5131
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
5132
TypeId: testComponentTypeID,
5133
},
5134
},
5135
},
5136
},
5137
}
5138
5139
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
5140
root, err := s.newTestTree(persistenceNodes)
5141
s.NoError(err)
5142
5143
t1 := s.timeSource.Now()
5144
t2 := t1.Add(time.Minute)
5145
5146
// First transaction: add an initial singleton pure task.
5147
mutableContext := NewMutableContext(context.Background(), root)
5148
c, err := root.Component(mutableContext, ComponentRef{})
5149
s.NoError(err)
5150
testComponent := c.(*TestComponent)
5151
5152
s.testLibrary.mockSingletonReplacePureTaskHandler.EXPECT().
5153
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
5154
mutableContext.AddTask(testComponent, TaskAttributes{ScheduledTime: t1}, &TestSingletonReplacePureTask{Data: []byte("first")})
5155
5156
mutation, err := root.CloseTransaction()
5157
s.NoError(err)
5158
rootAttr := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5159
s.Len(rootAttr.PureTasks, 1)
5160
s.Equal(testSingletonReplacePureTaskTypeID, rootAttr.PureTasks[0].TypeId)
5161
5162
// Second transaction: add a second singleton pure task with a different scheduled time.
5163
// closeTransactionCleanupInvalidTasks re-validates the existing task (1 call),
5164
// then closeTransactionHandleNewTasks validates the new task (1 call).
5165
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 3 }
5166
mutableContext = NewMutableContext(context.Background(), root)
5167
c, err = root.Component(mutableContext, ComponentRef{})
5168
s.NoError(err)
5169
testComponent = c.(*TestComponent)
5170
5171
s.testLibrary.mockSingletonReplacePureTaskHandler.EXPECT().
5172
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(2)
5173
mutableContext.AddTask(testComponent, TaskAttributes{ScheduledTime: t2}, &TestSingletonReplacePureTask{Data: []byte("second")})
5174
5175
mutation, err = root.CloseTransaction()
5176
s.NoError(err)
5177
rootAttr = mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5178
s.Len(rootAttr.PureTasks, 1, "replace mode must keep exactly one task")
5179
s.Equal(testSingletonReplacePureTaskTypeID, rootAttr.PureTasks[0].TypeId)
5180
s.Equal(t2.UTC(), rootAttr.PureTasks[0].ScheduledTime.AsTime(), "replace mode must use the new task's scheduled time")
5181
}
5182
5183
func (s *nodeSuite) TestCloseTransaction_SingletonTask_Ignore_Pure() {
5184
persistenceNodes := map[string]*persistencespb.ChasmNode{
5185
"": {
5186
Metadata: &persistencespb.ChasmNodeMetadata{
5187
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5188
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5189
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
5190
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
5191
TypeId: testComponentTypeID,
5192
},
5193
},
5194
},
5195
},
5196
}
5197
5198
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
5199
root, err := s.newTestTree(persistenceNodes)
5200
s.NoError(err)
5201
5202
t1 := s.timeSource.Now()
5203
t2 := t1.Add(time.Minute)
5204
5205
// First transaction: add the initial singleton pure task.
5206
mutableContext := NewMutableContext(context.Background(), root)
5207
c, err := root.Component(mutableContext, ComponentRef{})
5208
s.NoError(err)
5209
testComponent := c.(*TestComponent)
5210
5211
s.testLibrary.mockSingletonIgnorePureTaskHandler.EXPECT().
5212
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
5213
mutableContext.AddTask(testComponent, TaskAttributes{ScheduledTime: t1}, &TestSingletonIgnorePureTask{Data: []byte("first")})
5214
5215
mutation, err := root.CloseTransaction()
5216
s.NoError(err)
5217
rootAttr := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5218
s.Len(rootAttr.PureTasks, 1)
5219
firstTask := rootAttr.PureTasks[0]
5220
5221
// Second transaction: add a second singleton pure task — it should be discarded.
5222
// closeTransactionCleanupInvalidTasks re-validates the existing task (1 call),
5223
// then closeTransactionHandleNewTasks validates the new task (1 call).
5224
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 3 }
5225
mutableContext = NewMutableContext(context.Background(), root)
5226
c, err = root.Component(mutableContext, ComponentRef{})
5227
s.NoError(err)
5228
testComponent = c.(*TestComponent)
5229
5230
s.testLibrary.mockSingletonIgnorePureTaskHandler.EXPECT().
5231
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(2)
5232
mutableContext.AddTask(testComponent, TaskAttributes{ScheduledTime: t2}, &TestSingletonIgnorePureTask{Data: []byte("second")})
5233
5234
mutation, err = root.CloseTransaction()
5235
s.NoError(err)
5236
rootAttr = mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5237
s.Len(rootAttr.PureTasks, 1, "ignore mode must keep exactly one task")
5238
s.Equal(firstTask.ScheduledTime.AsTime(), rootAttr.PureTasks[0].ScheduledTime.AsTime(),
5239
"ignore mode must keep the original task's scheduled time")
5240
}
5241
5242
func (s *nodeSuite) TestCloseTransaction_SingletonTask_InvalidNewTask_DoesNotDisplaceExisting() {
5243
persistenceNodes := map[string]*persistencespb.ChasmNode{
5244
"": {
5245
Metadata: &persistencespb.ChasmNodeMetadata{
5246
InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5247
LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1},
5248
Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{
5249
ComponentAttributes: &persistencespb.ChasmComponentAttributes{
5250
TypeId: testComponentTypeID,
5251
},
5252
},
5253
},
5254
},
5255
}
5256
5257
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 2 }
5258
root, err := s.newTestTree(persistenceNodes)
5259
s.NoError(err)
5260
5261
// First transaction: add a valid singleton replace side-effect task.
5262
mutableContext := NewMutableContext(context.Background(), root)
5263
c, err := root.Component(mutableContext, ComponentRef{})
5264
s.NoError(err)
5265
testComponent := c.(*TestComponent)
5266
5267
s.testLibrary.mockSingletonReplaceSideEffectTaskHandler.EXPECT().
5268
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
5269
mutableContext.AddTask(testComponent, TaskAttributes{}, &TestSingletonReplaceSideEffectTask{Data: []byte("first")})
5270
5271
mutation, err := root.CloseTransaction()
5272
s.NoError(err)
5273
rootAttr := mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5274
s.Len(rootAttr.SideEffectTasks, 1)
5275
firstTask := rootAttr.SideEffectTasks[0]
5276
5277
// Second transaction: add an invalid singleton task — validation drops it before singleton logic runs,
5278
// so the existing task must be unaffected.
5279
// closeTransactionCleanupInvalidTasks re-validates the existing task first (returns true to keep it),
5280
// then closeTransactionHandleNewTasks validates the new task (returns false to drop it).
5281
s.nodeBackend.HandleNextTransitionCount = func() int64 { return 3 }
5282
mutableContext = NewMutableContext(context.Background(), root)
5283
c, err = root.Component(mutableContext, ComponentRef{})
5284
s.NoError(err)
5285
testComponent = c.(*TestComponent)
5286
5287
s.testLibrary.mockSingletonReplaceSideEffectTaskHandler.EXPECT().
5288
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(true, nil).Times(1)
5289
s.testLibrary.mockSingletonReplaceSideEffectTaskHandler.EXPECT().
5290
Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(false, nil).Times(1)
5291
mutableContext.AddTask(testComponent, TaskAttributes{}, &TestSingletonReplaceSideEffectTask{Data: []byte("invalid-second")})
5292
5293
mutation, err = root.CloseTransaction()
5294
s.NoError(err)
5295
rootAttr = mutation.UpdatedNodes[""].GetMetadata().GetComponentAttributes()
5296
s.Len(rootAttr.SideEffectTasks, 1, "invalid new task must not displace existing singleton")
5297
s.Equal(firstTask.VersionedTransition, rootAttr.SideEffectTasks[0].VersionedTransition,
5298
"existing task must be unchanged")
5299
}