go.temporal.io/server/chasm/field_test.go
437 LOC · 0 covered · 437 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package chasm
2
3
import (
4
"context"
5
"reflect"
6
"testing"
7
8
"github.com/stretchr/testify/require"
9
"github.com/stretchr/testify/suite"
10
persistencespb "go.temporal.io/server/api/persistence/v1"
11
"go.temporal.io/server/common/clock"
12
"go.temporal.io/server/common/definition"
13
"go.temporal.io/server/common/log"
14
"go.temporal.io/server/common/metrics"
15
"go.temporal.io/server/common/primitives"
16
"go.temporal.io/server/common/testing/protorequire"
17
"go.temporal.io/server/common/testing/testlogger"
18
"go.uber.org/mock/gomock"
19
)
20
21
type fieldSuite struct {
22
suite.Suite
23
*require.Assertions
24
protorequire.ProtoAssertions
25
26
controller *gomock.Controller
27
nodeBackend *MockNodeBackend
28
29
registry *Registry
30
timeSource *clock.EventTimeSource
31
nodePathEncoder NodePathEncoder
32
logger log.Logger
33
metricsHandler metrics.Handler
34
}
35
36
func TestFieldSuite(t *testing.T) {
37
suite.Run(t, new(fieldSuite))
38
}
39
40
func (s *fieldSuite) SetupTest() {
41
s.initAssertions()
42
s.controller = gomock.NewController(s.T())
43
s.nodeBackend = &MockNodeBackend{}
44
45
s.logger = testlogger.NewTestLogger(s.T(), testlogger.FailOnAnyUnexpectedError)
46
s.metricsHandler = metrics.NoopMetricsHandler
47
s.registry = NewRegistry(s.logger)
48
err := s.registry.Register(newTestLibrary(s.controller))
49
s.NoError(err)
50
51
s.timeSource = clock.NewEventTimeSource()
52
s.nodePathEncoder = &testNodePathEncoder{}
53
}
54
55
func (s *fieldSuite) SetupSubTest() {
56
s.initAssertions()
57
}
58
59
func (s *fieldSuite) initAssertions() {
60
// `s.Assertions` (as well as other test helpers which depends on `s.T()`) must be initialized on
61
// both test and subtest levels (but not suite level, where `s.T()` is `nil`).
62
//
63
// If these helpers are not reinitialized on subtest level, any failed `assert` in
64
// subtest will fail the entire test (not subtest) immediately without running other subtests.
65
66
s.Assertions = require.New(s.T())
67
s.ProtoAssertions = protorequire.New(s.T())
68
}
69
70
func (s *fieldSuite) TestInternalFieldName() {
71
fT := reflect.TypeFor[Field[any]]()
72
73
_, ok := fT.FieldByName(internalFieldName)
74
s.True(ok, "expected field %s not found", internalFieldName)
75
}
76
77
func (s *fieldSuite) TestFieldGetSimple() {
78
tests := []struct {
79
name string
80
field Field[*TestSubComponent1]
81
expected *TestSubComponent1
82
}{
83
{
84
name: "Get with non-nil value",
85
field: Field[*TestSubComponent1]{
86
Internal: newFieldInternalWithValue(
87
fieldTypeComponent,
88
&TestSubComponent1{SubComponent1Data: &protoMessageType{
89
CreateRequestId: "component-data",
90
}},
91
)},
92
expected: &TestSubComponent1{SubComponent1Data: &protoMessageType{
93
CreateRequestId: "component-data",
94
}},
95
},
96
{
97
name: "Get with nil value and nil node",
98
field: Field[*TestSubComponent1]{
99
Internal: newFieldInternalWithNode(nil),
100
},
101
expected: nil,
102
},
103
}
104
105
for _, tt := range tests {
106
s.Run(tt.name, func() {
107
result, _ := tt.field.TryGet(nil)
108
s.Equal(tt.expected, result)
109
})
110
}
111
}
112
113
func (s *fieldSuite) TestFieldGetComponent() {
114
serializedNodes := testComponentSerializedNodes()
115
116
node, err := s.newTestTree(serializedNodes)
117
s.NoError(err)
118
119
chasmContext := NewMutableContext(context.Background(), node)
120
121
c, err := node.Component(chasmContext, ComponentRef{componentPath: rootPath})
122
s.NoError(err)
123
s.NotNil(c)
124
125
tc := c.(*TestComponent)
126
127
sc1 := tc.SubComponent1.Get(chasmContext)
128
s.NotNil(sc1)
129
s.ProtoEqual(&protoMessageType{
130
CreateRequestId: "sub-component1-data",
131
}, sc1.SubComponent1Data)
132
133
sd1 := tc.SubData1.Get(chasmContext)
134
s.NotNil(sd1)
135
s.ProtoEqual(&protoMessageType{
136
CreateRequestId: "sub-data1",
137
}, sd1)
138
}
139
140
func (s *fieldSuite) newTestTree(
141
serializedNodes map[string]*persistencespb.ChasmNode,
142
) (*Node, error) {
143
if len(serializedNodes) == 0 {
144
return NewEmptyTree(
145
s.registry,
146
s.timeSource,
147
s.nodeBackend,
148
s.nodePathEncoder,
149
s.logger,
150
s.metricsHandler,
151
), nil
152
}
153
return NewTreeFromDB(
154
serializedNodes,
155
s.registry,
156
s.timeSource,
157
s.nodeBackend,
158
s.nodePathEncoder,
159
s.logger,
160
s.metricsHandler,
161
)
162
}
163
164
// setupComponentWithTree creates a basic component structure and attaches it to the tree.
165
func (s *fieldSuite) setupComponentWithTree(rootComponent *TestComponent) (*Node, MutableContext, error) {
166
rootNode := NewEmptyTree(
167
s.registry,
168
s.timeSource,
169
s.nodeBackend,
170
s.nodePathEncoder,
171
s.logger,
172
s.metricsHandler,
173
)
174
if err := rootNode.SetRootComponent(rootComponent); err != nil {
175
return nil, nil, err
176
}
177
178
return rootNode, NewMutableContext(context.Background(), rootNode), nil
179
}
180
181
func (s *fieldSuite) TestDeferredPointerResolution() {
182
workflowKey := definition.NewWorkflowKey(
183
primitives.NewUUID().String(),
184
primitives.NewUUID().String(),
185
primitives.NewUUID().String(),
186
)
187
s.nodeBackend = &MockNodeBackend{
188
HandleNextTransitionCount: func() int64 { return 1 },
189
HandleGetCurrentVersion: func() int64 { return 1 },
190
HandleGetWorkflowKey: func() definition.WorkflowKey { return workflowKey },
191
}
192
193
sc1 := &TestSubComponent1{
194
SubComponent1Data: &protoMessageType{
195
CreateRequestId: "sub-component1-data",
196
},
197
}
198
199
rootComponent := &TestComponent{
200
ComponentData: &protoMessageType{
201
CreateRequestId: "component-data",
202
},
203
SubComponent1: NewComponentField(nil, sc1),
204
}
205
206
rootNode, ctx, err := s.setupComponentWithTree(rootComponent)
207
s.NoError(err)
208
209
// Get components from tree to mark nodes as needing sync.
210
rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{})
211
s.NoError(err)
212
rootComponent = rootComponentInterface.(*TestComponent)
213
sc1 = rootComponent.SubComponent1.Get(ctx)
214
215
// sc1 (child) points to rootComponent (parent) via component pointer.
216
sc1.RootPointer = ComponentPointerTo(ctx, rootComponent)
217
218
// Verify deferred state.
219
s.Equal(fieldTypeDeferredPointer, sc1.RootPointer.Internal.fieldType())
220
s.Equal(rootComponent, sc1.RootPointer.Internal.v)
221
222
// CloseTransaction should resolve the deferred pointer.
223
mutations, err := rootNode.CloseTransaction()
224
s.NoError(err)
225
s.NotEmpty(mutations.UpdatedNodes)
226
227
// Verify the pointer was resolved to a regular pointer with path.
228
s.Equal(fieldTypePointer, sc1.RootPointer.Internal.fieldType())
229
230
cResolvedPath, ok := sc1.RootPointer.Internal.v.([]string)
231
s.True(ok)
232
s.Equal([]string{}, cResolvedPath)
233
234
// Verify we can dereference the component pointer.
235
resolvedComponent := sc1.RootPointer.Get(ctx)
236
s.Equal(rootComponent, resolvedComponent)
237
}
238
239
func (s *fieldSuite) TestMixedPointerScenario() {
240
workflowKey := definition.NewWorkflowKey(
241
primitives.NewUUID().String(),
242
primitives.NewUUID().String(),
243
primitives.NewUUID().String(),
244
)
245
s.nodeBackend = &MockNodeBackend{
246
HandleNextTransitionCount: func() int64 { return 1 },
247
HandleGetCurrentVersion: func() int64 { return 1 },
248
HandleGetWorkflowKey: func() definition.WorkflowKey { return workflowKey },
249
}
250
251
sc11 := &TestSubComponent11{
252
SubComponent11Data: &protoMessageType{CreateRequestId: "sub-component11-data"},
253
}
254
255
sc1 := &TestSubComponent1{
256
SubComponent1Data: &protoMessageType{CreateRequestId: "sub-component1-data"},
257
SubComponent11: NewComponentField(nil, sc11),
258
}
259
260
rootComponent := &TestComponent{
261
ComponentData: &protoMessageType{CreateRequestId: "component-data"},
262
SubComponent1: NewComponentField(nil, sc1),
263
}
264
265
rootNode, ctx, err := s.setupComponentWithTree(rootComponent)
266
s.NoError(err)
267
268
// Get components from tree to mark nodes as needing sync.
269
rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{})
270
s.NoError(err)
271
rootComponent = rootComponentInterface.(*TestComponent)
272
sc1 = rootComponent.SubComponent1.Get(ctx)
273
sc11 = sc1.SubComponent11.Get(ctx)
274
275
// Transaction 1: sc11 points to root (grandparent).
276
sc11.GrandparentPointer = ComponentPointerTo(ctx, rootComponent)
277
278
_, err = rootNode.CloseTransaction()
279
s.NoError(err)
280
s.Equal(fieldTypePointer, sc11.GrandparentPointer.Internal.fieldType())
281
282
// Transaction 2: sc1 points to root (parent).
283
ctx2 := NewMutableContext(context.Background(), rootNode)
284
rootComponentInterface, err = rootNode.Component(ctx2, ComponentRef{})
285
s.NoError(err)
286
287
rootComponent = rootComponentInterface.(*TestComponent)
288
sc1 = rootComponent.SubComponent1.Get(ctx2)
289
sc11 = sc1.SubComponent11.Get(ctx2)
290
291
sc1.RootPointer = ComponentPointerTo(ctx2, rootComponent)
292
293
s.Equal(fieldTypePointer, sc11.GrandparentPointer.Internal.fieldType())
294
s.Equal(fieldTypeDeferredPointer, sc1.RootPointer.Internal.fieldType())
295
296
_, err = rootNode.CloseTransaction()
297
s.NoError(err)
298
299
// Ensure both pointers have been resolved.
300
s.Equal(fieldTypePointer, sc11.GrandparentPointer.Internal.fieldType())
301
s.Equal(fieldTypePointer, sc1.RootPointer.Internal.fieldType())
302
303
resolved1 := sc11.GrandparentPointer.Get(ctx2)
304
s.Equal(rootComponent, resolved1)
305
306
resolved2 := sc1.RootPointer.Get(ctx2)
307
s.Equal(rootComponent, resolved2)
308
}
309
310
func (s *fieldSuite) TestUnresolvableDeferredPointerError() {
311
workflowKey := definition.NewWorkflowKey(
312
primitives.NewUUID().String(),
313
primitives.NewUUID().String(),
314
primitives.NewUUID().String(),
315
)
316
s.nodeBackend = &MockNodeBackend{
317
HandleNextTransitionCount: func() int64 { return 1 },
318
HandleGetCurrentVersion: func() int64 { return 1 },
319
HandleGetWorkflowKey: func() definition.WorkflowKey { return workflowKey },
320
}
321
322
s.logger.(*testlogger.TestLogger).
323
Expect(testlogger.Error, "failed to resolve deferred pointer during transaction close")
324
325
orphanComponent := &TestSubComponent11{
326
SubComponent11Data: &protoMessageType{
327
CreateRequestId: "orphan-component",
328
},
329
}
330
331
rootComponent := &TestComponent{
332
ComponentData: &protoMessageType{
333
CreateRequestId: "component-data",
334
},
335
}
336
337
rootNode, ctx, err := s.setupComponentWithTree(rootComponent)
338
s.NoError(err)
339
340
// Get component from tree to mark node as needing sync.
341
rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{})
342
s.NoError(err)
343
rootComponent = rootComponentInterface.(*TestComponent)
344
345
rootComponent.SubComponent11Pointer = ComponentPointerTo(ctx, orphanComponent)
346
s.Equal(fieldTypeDeferredPointer, rootComponent.SubComponent11Pointer.Internal.fieldType())
347
348
_, err = rootNode.CloseTransaction()
349
s.Error(err)
350
s.Contains(err.Error(), "failed to resolve deferred pointer during transaction close")
351
}
352
353
func (s *fieldSuite) TestNonAncestorComponentPointerRejected() {
354
workflowKey := definition.NewWorkflowKey(
355
primitives.NewUUID().String(),
356
primitives.NewUUID().String(),
357
primitives.NewUUID().String(),
358
)
359
s.nodeBackend = &MockNodeBackend{
360
HandleNextTransitionCount: func() int64 { return 1 },
361
HandleGetCurrentVersion: func() int64 { return 1 },
362
HandleGetWorkflowKey: func() definition.WorkflowKey { return workflowKey },
363
}
364
365
s.logger.(*testlogger.TestLogger).
366
Expect(testlogger.Error, "failed to resolve deferred pointer during transaction close")
367
368
sc11 := &TestSubComponent11{
369
SubComponent11Data: &protoMessageType{CreateRequestId: "sub-component11-data"},
370
}
371
372
sc1 := &TestSubComponent1{
373
SubComponent1Data: &protoMessageType{CreateRequestId: "sub-component1-data"},
374
SubComponent11: NewComponentField(nil, sc11),
375
}
376
377
rootComponent := &TestComponent{
378
ComponentData: &protoMessageType{CreateRequestId: "component-data"},
379
SubComponent1: NewComponentField(nil, sc1),
380
}
381
382
rootNode, ctx, err := s.setupComponentWithTree(rootComponent)
383
s.NoError(err)
384
385
rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{})
386
s.NoError(err)
387
rootComponent = rootComponentInterface.(*TestComponent)
388
sc1 = rootComponent.SubComponent1.Get(ctx)
389
sc11 = sc1.SubComponent11.Get(ctx)
390
391
// Root pointing to descendant sc11 should be rejected.
392
rootComponent.SubComponent11Pointer = ComponentPointerTo(ctx, sc11)
393
394
_, err = rootNode.CloseTransaction()
395
s.Error(err)
396
s.Contains(err.Error(), "is not an ancestor of component")
397
}
398
399
func (s *fieldSuite) TestChildComponentPointerRejected() {
400
workflowKey := definition.NewWorkflowKey(
401
primitives.NewUUID().String(),
402
primitives.NewUUID().String(),
403
primitives.NewUUID().String(),
404
)
405
s.nodeBackend = &MockNodeBackend{
406
HandleNextTransitionCount: func() int64 { return 1 },
407
HandleGetCurrentVersion: func() int64 { return 1 },
408
HandleGetWorkflowKey: func() definition.WorkflowKey { return workflowKey },
409
}
410
411
s.logger.(*testlogger.TestLogger).
412
Expect(testlogger.Error, "failed to resolve deferred pointer during transaction close")
413
414
sc1 := &TestSubComponent1{
415
SubComponent1Data: &protoMessageType{CreateRequestId: "sub-component1-data"},
416
}
417
418
rootComponent := &TestComponent{
419
ComponentData: &protoMessageType{CreateRequestId: "component-data"},
420
SubComponent1: NewComponentField(nil, sc1),
421
}
422
423
rootNode, ctx, err := s.setupComponentWithTree(rootComponent)
424
s.NoError(err)
425
426
rootComponentInterface, err := rootNode.Component(ctx, ComponentRef{})
427
s.NoError(err)
428
rootComponent = rootComponentInterface.(*TestComponent)
429
sc1 = rootComponent.SubComponent1.Get(ctx)
430
431
// Root pointing to child sc1 via interface pointer should be rejected.
432
rootComponent.SubComponentInterfacePointer = ComponentPointerTo[Component](ctx, sc1)
433
434
_, err = rootNode.CloseTransaction()
435
s.Error(err)
436
s.Contains(err.Error(), "is not an ancestor of component")
437
}