go.temporal.io/server/tests/chasm_test.go
1123 LOC · 0 covered · 1123 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package tests
2
3
import (
4
"context"
5
"crypto/rand"
6
"errors"
7
"fmt"
8
"strconv"
9
"testing"
10
"time"
11
12
"github.com/stretchr/testify/require"
13
commonpb "go.temporal.io/api/common/v1"
14
enumspb "go.temporal.io/api/enums/v1"
15
"go.temporal.io/api/operatorservice/v1"
16
"go.temporal.io/api/serviceerror"
17
workflowpb "go.temporal.io/api/workflow/v1"
18
"go.temporal.io/api/workflowservice/v1"
19
"go.temporal.io/server/api/adminservice/v1"
20
"go.temporal.io/server/chasm"
21
"go.temporal.io/server/chasm/lib/tests"
22
testspb "go.temporal.io/server/chasm/lib/tests/gen/testspb/v1"
23
"go.temporal.io/server/common/debug"
24
"go.temporal.io/server/common/dynamicconfig"
25
"go.temporal.io/server/common/namespace"
26
"go.temporal.io/server/common/payload"
27
"go.temporal.io/server/common/searchattribute/sadefs"
28
"go.temporal.io/server/common/testing/await"
29
"go.temporal.io/server/common/testing/parallelsuite"
30
"go.temporal.io/server/common/testing/testvars"
31
"go.temporal.io/server/tests/testcore"
32
"google.golang.org/protobuf/types/known/durationpb"
33
)
34
35
const (
36
chasmTestTimeout = 10 * time.Second * debug.TimeoutMultiplier
37
)
38
39
// ChasmSuite runs CHASM functional tests using a pooled dedicated cluster.
40
// Each test is exercised with both the legacy and unified visibility query converters
41
// via s.Run subtests named "unified=false" and "unified=true".
42
type ChasmSuite struct {
43
parallelsuite.Suite[*ChasmSuite]
44
}
45
46
func TestChasmSuite(t *testing.T) {
47
parallelsuite.Run(t, &ChasmSuite{})
48
}
49
50
// chasmTestEnv bundles a TestEnv with the CHASM engine context derived from it.
51
type chasmTestEnv struct {
52
*testcore.TestEnv
53
chasmCtx context.Context
54
}
55
56
// newChasmTestEnv creates a chasmTestEnv backed by a dedicated cluster with
57
// EnableChasm and VisibilityEnableUnifiedQueryConverter overridden for the test.
58
func newChasmTestEnv(suiteContext func() context.Context, t *testing.T, unified bool) chasmTestEnv {
59
t.Helper()
60
61
// WithDedicatedCluster acquires an exclusive pooled slot — no fresh cluster
62
// creation per test, unlike passing startup dynamic config.
63
env := testcore.NewEnv(
64
t,
65
testcore.WithDedicatedCluster(),
66
testcore.WithWorkerService("delete namespace workflow"),
67
testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true),
68
testcore.WithDynamicConfig(dynamicconfig.VisibilityEnableUnifiedQueryConverter, unified),
69
testcore.WithDynamicConfig(dynamicconfig.DeleteNamespaceUseChasmDeleteExecution, true),
70
)
71
72
chasmCtx, err := env.GetTestCluster().Host().ChasmContext(suiteContext())
73
require.NoError(t, err)
74
75
return chasmTestEnv{TestEnv: env, chasmCtx: chasmCtx}
76
}
77
78
// forBothConverters runs fn as two parallel subtests, one with the legacy visibility
79
// query converter and one with the unified converter.
80
// TODO: Remove once we have fully migrated to the unified query converter.
81
func (s *ChasmSuite) forBothConverters(fn func(*ChasmSuite, chasmTestEnv)) {
82
for _, unified := range []bool{false, true} {
83
s.Run(fmt.Sprintf("unified=%v", unified), func(ss *ChasmSuite) {
84
// Resolve the suite context after NewEnv attaches its RPC header decorator.
85
fn(ss, newChasmTestEnv(ss.Context, ss.T(), unified))
86
})
87
}
88
}
89
90
func (s *ChasmSuite) TestNewPayloadStore() {
91
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
92
tv := testvars.New(ss.T())
93
94
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
95
defer cancel()
96
97
_, err := tests.NewPayloadStoreHandler(
98
ctx,
99
tests.NewPayloadStoreRequest{
100
NamespaceID: cenv.NamespaceID(),
101
StoreID: tv.Any().String(),
102
IDReusePolicy: chasm.BusinessIDReusePolicyRejectDuplicate,
103
IDConflictPolicy: chasm.BusinessIDConflictPolicyFail,
104
},
105
)
106
ss.NoError(err)
107
})
108
}
109
110
func (s *ChasmSuite) TestNewPayloadStore_ConflictPolicy_UseExisting() {
111
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
112
tv := testvars.New(ss.T())
113
114
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
115
defer cancel()
116
117
storeID := tv.Any().String()
118
119
resp, err := tests.NewPayloadStoreHandler(
120
ctx,
121
tests.NewPayloadStoreRequest{
122
NamespaceID: cenv.NamespaceID(),
123
StoreID: storeID,
124
IDReusePolicy: chasm.BusinessIDReusePolicyRejectDuplicate,
125
IDConflictPolicy: chasm.BusinessIDConflictPolicyFail,
126
},
127
)
128
ss.NoError(err)
129
130
currentRunID := resp.RunID
131
132
resp, err = tests.NewPayloadStoreHandler(
133
ctx,
134
tests.NewPayloadStoreRequest{
135
NamespaceID: cenv.NamespaceID(),
136
StoreID: storeID,
137
IDReusePolicy: chasm.BusinessIDReusePolicyRejectDuplicate,
138
IDConflictPolicy: chasm.BusinessIDConflictPolicyFail,
139
},
140
)
141
ss.ErrorAs(err, new(*chasm.ExecutionAlreadyStartedError))
142
143
resp, err = tests.NewPayloadStoreHandler(
144
ctx,
145
tests.NewPayloadStoreRequest{
146
NamespaceID: cenv.NamespaceID(),
147
StoreID: storeID,
148
IDReusePolicy: chasm.BusinessIDReusePolicyRejectDuplicate,
149
IDConflictPolicy: chasm.BusinessIDConflictPolicyUseExisting,
150
},
151
)
152
ss.NoError(err)
153
ss.Equal(currentRunID, resp.RunID)
154
})
155
}
156
157
func (s *ChasmSuite) TestPayloadStore_UpdateComponent() {
158
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
159
tv := testvars.New(ss.T())
160
161
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
162
defer cancel()
163
164
storeID := tv.Any().String()
165
_, err := tests.NewPayloadStoreHandler(
166
ctx,
167
tests.NewPayloadStoreRequest{
168
NamespaceID: cenv.NamespaceID(),
169
StoreID: storeID,
170
},
171
)
172
ss.NoError(err)
173
174
_, err = tests.AddPayloadHandler(
175
ctx,
176
tests.AddPayloadRequest{
177
NamespaceID: cenv.NamespaceID(),
178
StoreID: storeID,
179
PayloadKey: "key1",
180
Payload: payload.EncodeString("value1"),
181
},
182
)
183
ss.NoError(err)
184
185
descResp, err := tests.DescribePayloadStoreHandler(
186
ctx,
187
tests.DescribePayloadStoreRequest{
188
NamespaceID: cenv.NamespaceID(),
189
StoreID: storeID,
190
},
191
)
192
ss.NoError(err)
193
ss.Equal(int64(1), descResp.State.TotalCount)
194
ss.Positive(descResp.State.TotalSize)
195
})
196
}
197
198
func (s *ChasmSuite) TestPayloadStore_PureTask() {
199
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
200
tv := testvars.New(ss.T())
201
202
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
203
defer cancel()
204
205
storeID := tv.Any().String()
206
_, err := tests.NewPayloadStoreHandler(
207
ctx,
208
tests.NewPayloadStoreRequest{
209
NamespaceID: cenv.NamespaceID(),
210
StoreID: storeID,
211
},
212
)
213
ss.NoError(err)
214
215
_, err = tests.AddPayloadHandler(
216
ctx,
217
tests.AddPayloadRequest{
218
NamespaceID: cenv.NamespaceID(),
219
StoreID: storeID,
220
PayloadKey: "key1",
221
Payload: payload.EncodeString("value1"),
222
TTL: 1 * time.Second,
223
},
224
)
225
ss.NoError(err)
226
227
ss.AwaitTrue(func() bool {
228
descResp, err := tests.DescribePayloadStoreHandler(
229
ctx,
230
tests.DescribePayloadStoreRequest{
231
NamespaceID: cenv.NamespaceID(),
232
StoreID: storeID,
233
},
234
)
235
ss.NoError(err)
236
return descResp.State.TotalCount == 0
237
}, 10*time.Second, 100*time.Millisecond)
238
})
239
}
240
241
func (s *ChasmSuite) TestListExecutions() {
242
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
243
tv := testvars.New(ss.T())
244
245
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
246
defer cancel()
247
248
storeID := tv.Any().String()
249
createResp, err := tests.NewPayloadStoreHandler(
250
ctx,
251
tests.NewPayloadStoreRequest{
252
NamespaceID: cenv.NamespaceID(),
253
StoreID: storeID,
254
},
255
)
256
ss.NoError(err)
257
258
visQuery := fmt.Sprintf("TemporalNamespaceDivision = '%d' AND PayloadStoreId = '%s'", tests.ArchetypeID, storeID)
259
260
var visRecord *chasm.VisibilityExecutionInfo[*testspb.TestPayloadStore]
261
ss.AwaitTrue(
262
func() bool {
263
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
264
NamespaceName: string(cenv.Namespace()),
265
PageSize: 10,
266
Query: visQuery,
267
})
268
ss.NoError(err)
269
if len(resp.Executions) != 1 {
270
return false
271
}
272
273
visRecord = resp.Executions[0]
274
return true
275
},
276
testcore.WaitForESToSettle,
277
100*time.Millisecond,
278
)
279
ss.Equal(storeID, visRecord.BusinessID)
280
ss.Equal(createResp.RunID, visRecord.RunID)
281
ss.NotEmpty(visRecord.StartTime)
282
ss.Empty(visRecord.StateTransitionCount)
283
284
totalCount := visRecord.ChasmMemo.TotalCount
285
ss.Equal(0, int(totalCount))
286
totalSize := visRecord.ChasmMemo.TotalSize
287
ss.Equal(0, int(totalSize))
288
totalCountSA, ok := chasm.SearchAttributeValue(visRecord.ChasmSearchAttributes, tests.PayloadTotalCountSearchAttribute)
289
ss.True(ok)
290
ss.Equal(0, int(totalCountSA))
291
totalSizeSA, ok := chasm.SearchAttributeValue(visRecord.ChasmSearchAttributes, tests.PayloadTotalSizeSearchAttribute)
292
ss.True(ok)
293
ss.Equal(0, int(totalSizeSA))
294
var scheduledByID string
295
ss.NoError(payload.Decode(visRecord.CustomSearchAttributes[sadefs.TemporalScheduledById], &scheduledByID))
296
ss.Equal(tests.TestScheduleID, scheduledByID)
297
var archetypeIDStr string
298
ss.NoError(payload.Decode(visRecord.CustomSearchAttributes[sadefs.TemporalNamespaceDivision], &archetypeIDStr))
299
parsedArchetypeID, err := strconv.ParseUint(archetypeIDStr, 10, 32)
300
ss.NoError(err)
301
ss.Equal(tests.ArchetypeID, chasm.ArchetypeID(parsedArchetypeID))
302
303
addPayloadResp, err := tests.AddPayloadHandler(
304
ctx,
305
tests.AddPayloadRequest{
306
NamespaceID: cenv.NamespaceID(),
307
StoreID: storeID,
308
PayloadKey: "key1",
309
Payload: payload.EncodeString("value1"),
310
},
311
)
312
ss.NoError(err)
313
314
ss.AwaitTrue(
315
func() bool {
316
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
317
NamespaceName: string(cenv.Namespace()),
318
PageSize: 10,
319
Query: visQuery + " AND PayloadTotalCount > 0",
320
})
321
ss.NoError(err)
322
if len(resp.Executions) != 1 {
323
return false
324
}
325
326
visRecord = resp.Executions[0]
327
return visRecord.ChasmMemo.TotalCount == addPayloadResp.State.TotalCount
328
},
329
testcore.WaitForESToSettle,
330
100*time.Millisecond,
331
)
332
// We validated Count memo field above, just checking for size here.
333
ss.Equal(addPayloadResp.State.TotalSize, visRecord.ChasmMemo.TotalSize)
334
335
_, err = tests.ClosePayloadStoreHandler(
336
ctx,
337
tests.ClosePayloadStoreRequest{
338
NamespaceID: cenv.NamespaceID(),
339
StoreID: storeID,
340
},
341
)
342
ss.NoError(err)
343
344
ss.AwaitTrue(
345
func() bool {
346
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
347
NamespaceName: cenv.Namespace().String(),
348
PageSize: 10,
349
Query: visQuery + " AND ExecutionStatus = 'Completed' AND PayloadTotalCount > 0",
350
})
351
ss.NoError(err)
352
if len(resp.Executions) != 1 {
353
return false
354
}
355
356
visRecord = resp.Executions[0]
357
return true
358
},
359
testcore.WaitForESToSettle,
360
100*time.Millisecond,
361
)
362
ss.Equal(int64(3), visRecord.StateTransitionCount)
363
ss.NotEmpty(visRecord.CloseTime)
364
ss.Empty(visRecord.HistoryLength)
365
})
366
}
367
368
func (s *ChasmSuite) TestCountExecutions_GroupBy() {
369
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
370
tv := testvars.New(ss.T())
371
372
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
373
defer cancel()
374
375
for range 3 {
376
storeID := tv.Any().String()
377
378
_, err := tests.NewPayloadStoreHandler(
379
cenv.chasmCtx,
380
tests.NewPayloadStoreRequest{
381
NamespaceID: cenv.NamespaceID(),
382
StoreID: storeID,
383
},
384
)
385
ss.NoError(err)
386
}
387
388
for range 2 {
389
storeID := tv.Any().String()
390
391
resp, err := tests.NewPayloadStoreHandler(
392
cenv.chasmCtx,
393
tests.NewPayloadStoreRequest{
394
NamespaceID: cenv.NamespaceID(),
395
StoreID: storeID,
396
},
397
)
398
ss.NoError(err)
399
400
_, err = tests.AddPayloadHandler(
401
cenv.chasmCtx,
402
tests.AddPayloadRequest{
403
NamespaceID: cenv.NamespaceID(),
404
StoreID: storeID,
405
PayloadKey: "key1",
406
Payload: payload.EncodeString("value1"),
407
},
408
)
409
ss.NoError(err)
410
411
_, err = tests.CancelPayloadStoreHandler(
412
cenv.chasmCtx,
413
tests.CancelPayloadStoreRequest{
414
NamespaceID: cenv.NamespaceID(),
415
StoreID: storeID,
416
},
417
)
418
ss.NoError(err)
419
ss.NotEmpty(resp.RunID)
420
}
421
422
var countResp *chasm.CountExecutionsResponse
423
var err error
424
ss.AwaitTrue(
425
func() bool {
426
countResp, err = chasm.CountExecutions[*tests.PayloadStore](
427
ctx,
428
&chasm.CountExecutionsRequest{
429
NamespaceName: cenv.Namespace().String(),
430
Query: "GROUP BY `ExecutionStatus`",
431
},
432
)
433
return err == nil && countResp != nil && countResp.Count >= 5
434
},
435
testcore.WaitForESToSettle,
436
100*time.Millisecond,
437
)
438
439
ss.NoError(err)
440
ss.NotNil(countResp)
441
ss.Equal(int64(5), countResp.Count)
442
ss.Len(countResp.Groups, 2)
443
444
var totalCount int64
445
for _, group := range countResp.Groups {
446
ss.Len(group.Values, 1)
447
totalCount += group.Count
448
var groupValue string
449
ss.NoError(payload.Decode(group.Values[0], &groupValue))
450
ss.Contains([]string{"Running", "Canceled"}, groupValue)
451
}
452
ss.Equal(int64(5), totalCount)
453
454
// Test that GROUP BY on unsupported field returns error
455
_, err = chasm.CountExecutions[*tests.PayloadStore](
456
ctx,
457
&chasm.CountExecutionsRequest{
458
NamespaceName: cenv.Namespace().String(),
459
Query: "GROUP BY `PayloadTotalCount`",
460
},
461
)
462
var invalidArgument *serviceerror.InvalidArgument
463
ss.ErrorAs(err, &invalidArgument)
464
ss.Contains(err.Error(), "GROUP BY")
465
})
466
}
467
468
func (s *ChasmSuite) TestListWorkflowExecutions() {
469
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
470
tv := testvars.New(ss.T())
471
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
472
defer cancel()
473
474
storeID := tv.Any().String()
475
createResp, err := tests.NewPayloadStoreHandler(
476
ctx,
477
tests.NewPayloadStoreRequest{
478
NamespaceID: cenv.NamespaceID(),
479
StoreID: storeID,
480
},
481
)
482
ss.NoError(err)
483
484
_, err = tests.AddPayloadHandler(
485
ctx,
486
tests.AddPayloadRequest{
487
NamespaceID: cenv.NamespaceID(),
488
StoreID: storeID,
489
PayloadKey: "test-key",
490
Payload: payload.EncodeString("test-value"),
491
},
492
)
493
ss.NoError(err)
494
495
visQuery := sadefs.QueryWithAnyNamespaceDivision(
496
fmt.Sprintf("WorkflowId = '%s'", storeID),
497
)
498
499
var execInfo *workflowpb.WorkflowExecutionInfo
500
ss.AwaitTrue(
501
func() bool {
502
listResp, err := cenv.FrontendClient().ListWorkflowExecutions(ss.Context(), &workflowservice.ListWorkflowExecutionsRequest{
503
Namespace: cenv.Namespace().String(),
504
PageSize: 10,
505
Query: visQuery,
506
})
507
ss.NoError(err)
508
if len(listResp.Executions) != 1 {
509
return false
510
}
511
execInfo = listResp.Executions[0]
512
return true
513
},
514
testcore.WaitForESToSettle,
515
100*time.Millisecond,
516
)
517
518
ss.Equal(storeID, execInfo.Execution.WorkflowId)
519
ss.Equal(createResp.RunID, execInfo.Execution.RunId)
520
521
ss.NotNil(execInfo.SearchAttributes)
522
_, hasScheduledByID := execInfo.SearchAttributes.IndexedFields[sadefs.TemporalScheduledById]
523
ss.True(hasScheduledByID)
524
525
_, hasTotalCount := execInfo.SearchAttributes.IndexedFields["TemporalInt01"]
526
ss.False(hasTotalCount, "CHASM search attribute TemporalInt01 should not be exposed")
527
_, hasTotalSize := execInfo.SearchAttributes.IndexedFields["TemporalInt02"]
528
ss.False(hasTotalSize, "CHASM search attribute TemporalInt02 should not be exposed")
529
})
530
}
531
532
func (s *ChasmSuite) TestPayloadStoreForceDelete() {
533
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
534
tv := testvars.New(ss.T())
535
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
536
defer cancel()
537
538
storeID := tv.Any().String()
539
createResp, err := tests.NewPayloadStoreHandler(
540
ctx,
541
tests.NewPayloadStoreRequest{
542
NamespaceID: cenv.NamespaceID(),
543
StoreID: storeID,
544
IDReusePolicy: chasm.BusinessIDReusePolicyRejectDuplicate,
545
IDConflictPolicy: chasm.BusinessIDConflictPolicyFail,
546
},
547
)
548
ss.NoError(err)
549
550
// Make sure visibility record is created, so that we can test its deletion later.
551
visQuery := fmt.Sprintf("TemporalNamespaceDivision = '%d' AND WorkflowId = '%s'", tests.ArchetypeID, storeID)
552
var executionInfo *workflowpb.WorkflowExecutionInfo
553
ss.AwaitTrue(
554
func() bool {
555
resp, err := cenv.FrontendClient().ListWorkflowExecutions(ss.Context(), &workflowservice.ListWorkflowExecutionsRequest{
556
Namespace: cenv.Namespace().String(),
557
PageSize: 10,
558
Query: visQuery,
559
})
560
ss.NoError(err)
561
if len(resp.Executions) > 0 {
562
executionInfo = resp.Executions[0]
563
}
564
return len(resp.Executions) == 1
565
},
566
testcore.WaitForESToSettle,
567
100*time.Millisecond,
568
)
569
archetypePayload, ok := executionInfo.SearchAttributes.GetIndexedFields()[sadefs.TemporalNamespaceDivision]
570
ss.True(ok)
571
var archetypeIDStr string
572
ss.NoError(payload.Decode(archetypePayload, &archetypeIDStr))
573
parsedArchetypeID, err := strconv.ParseUint(archetypeIDStr, 10, 32)
574
ss.NoError(err)
575
ss.Equal(tests.ArchetypeID, chasm.ArchetypeID(parsedArchetypeID))
576
577
_, err = cenv.AdminClient().DeleteWorkflowExecution(ss.Context(), &adminservice.DeleteWorkflowExecutionRequest{
578
Namespace: cenv.Namespace().String(),
579
Execution: &commonpb.WorkflowExecution{
580
WorkflowId: storeID,
581
RunId: createResp.RunID,
582
},
583
Archetype: tests.Archetype,
584
})
585
ss.NoError(err)
586
587
// Validate mutable state is deleted.
588
_, err = cenv.AdminClient().DescribeMutableState(ss.Context(), &adminservice.DescribeMutableStateRequest{
589
Namespace: cenv.Namespace().String(),
590
Execution: &commonpb.WorkflowExecution{
591
WorkflowId: storeID,
592
RunId: createResp.RunID,
593
},
594
Archetype: tests.Archetype,
595
})
596
var notFoundErr *serviceerror.NotFound
597
ss.ErrorAs(err, ¬FoundErr)
598
599
// Validate visibility record is deleted.
600
ss.AwaitTrue(
601
func() bool {
602
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
603
NamespaceName: cenv.Namespace().String(),
604
PageSize: 10,
605
Query: visQuery,
606
})
607
ss.NoError(err)
608
return len(resp.Executions) == 0
609
},
610
testcore.WaitForESToSettle,
611
100*time.Millisecond,
612
)
613
})
614
}
615
616
func (s *ChasmSuite) TestDeletePayloadStore_RunningExecution() {
617
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
618
tv := testvars.New(ss.T())
619
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
620
defer cancel()
621
622
storeID := tv.Any().String()
623
_, err := tests.NewPayloadStoreHandler(
624
ctx,
625
tests.NewPayloadStoreRequest{
626
NamespaceID: cenv.NamespaceID(),
627
StoreID: storeID,
628
IDReusePolicy: chasm.BusinessIDReusePolicyRejectDuplicate,
629
IDConflictPolicy: chasm.BusinessIDConflictPolicyFail,
630
},
631
)
632
ss.NoError(err)
633
634
visQuery := fmt.Sprintf("WorkflowId = '%s'", storeID)
635
636
// Wait for visibility record to appear.
637
ss.Await(
638
func(ss *ChasmSuite) {
639
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
640
NamespaceName: cenv.Namespace().String(),
641
PageSize: 10,
642
Query: visQuery,
643
})
644
ss.NoError(err)
645
ss.Len(resp.Executions, 1)
646
},
647
testcore.WaitForESToSettle,
648
100*time.Millisecond,
649
)
650
651
err = tests.DeletePayloadStoreHandler(
652
ctx,
653
tests.DeletePayloadStoreRequest{
654
NamespaceID: cenv.NamespaceID(),
655
StoreID: storeID,
656
Reason: "test deletion",
657
Identity: "test-identity",
658
},
659
)
660
ss.NoError(err)
661
662
// Validate execution is fully deleted (both mutable state and visibility record).
663
ss.Await(
664
func(ss *ChasmSuite) {
665
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
666
NamespaceName: cenv.Namespace().String(),
667
PageSize: 10,
668
Query: visQuery,
669
})
670
ss.NoError(err)
671
ss.Empty(resp.Executions)
672
},
673
testcore.WaitForESToSettle,
674
100*time.Millisecond,
675
)
676
})
677
}
678
679
func (s *ChasmSuite) TestListExecutions_ExecutionStatusAsAlias() {
680
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
681
tv := testvars.New(ss.T())
682
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
683
defer cancel()
684
685
storeID := tv.Any().String()
686
_, err := tests.NewPayloadStoreHandler(
687
ctx,
688
tests.NewPayloadStoreRequest{
689
NamespaceID: cenv.NamespaceID(),
690
StoreID: storeID,
691
},
692
)
693
ss.NoError(err)
694
695
// Query using "ExecutionStatus" as a CHASM alias (which maps to TemporalKeyword03).
696
// This tests that CHASM components can use "ExecutionStatus" as an alias for their own search attribute.
697
visQuery := fmt.Sprintf("TemporalNamespaceDivision = '%d' AND ExecutionStatus = 'Running' AND PayloadStoreId = '%s'", tests.ArchetypeID, storeID)
698
699
var visRecord *chasm.VisibilityExecutionInfo[*testspb.TestPayloadStore]
700
ss.AwaitTrue(
701
func() bool {
702
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
703
NamespaceName: string(cenv.Namespace()),
704
PageSize: 10,
705
Query: visQuery,
706
})
707
ss.NoError(err)
708
if len(resp.Executions) != 1 {
709
return false
710
}
711
712
visRecord = resp.Executions[0]
713
return true
714
},
715
testcore.WaitForESToSettle,
716
100*time.Millisecond,
717
)
718
ss.Equal(storeID, visRecord.BusinessID)
719
720
// Verify the ExecutionStatus CHASM search attribute is correctly returned.
721
executionStatus, ok := chasm.SearchAttributeValue(visRecord.ChasmSearchAttributes, tests.ExecutionStatusSearchAttribute)
722
ss.True(ok)
723
ss.Equal("Running", executionStatus)
724
725
_, err = tests.CancelPayloadStoreHandler(
726
ctx,
727
tests.CancelPayloadStoreRequest{
728
NamespaceID: cenv.NamespaceID(),
729
StoreID: storeID,
730
},
731
)
732
ss.NoError(err)
733
734
visQueryCanceled := fmt.Sprintf("TemporalNamespaceDivision = '%d' AND ExecutionStatus = 'Canceled' AND PayloadStoreId = '%s'", tests.ArchetypeID, storeID)
735
ss.AwaitTrue(
736
func() bool {
737
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
738
NamespaceName: string(cenv.Namespace()),
739
PageSize: 10,
740
Query: visQueryCanceled,
741
})
742
ss.NoError(err)
743
return len(resp.Executions) == 1
744
},
745
testcore.WaitForESToSettle,
746
100*time.Millisecond,
747
)
748
})
749
}
750
751
func (s *ChasmSuite) TestTaskQueuePreallocatedSearchAttribute() {
752
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
753
tv := testvars.New(ss.T())
754
755
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
756
defer cancel()
757
758
storeID := tv.Any().String()
759
760
_, err := tests.NewPayloadStoreHandler(
761
ctx,
762
tests.NewPayloadStoreRequest{
763
NamespaceID: cenv.NamespaceID(),
764
StoreID: storeID,
765
},
766
)
767
ss.NoError(err)
768
769
// Query using TaskQueue as a CHASM preallocated search attribute.
770
visQuery := fmt.Sprintf("TemporalNamespaceDivision = '%d' AND TaskQueue = '%s' AND PayloadStoreId = '%s'", tests.ArchetypeID, tests.DefaultPayloadStoreTaskQueue, storeID)
771
772
var visRecord *chasm.VisibilityExecutionInfo[*testspb.TestPayloadStore]
773
ss.AwaitTrue(
774
func() bool {
775
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
776
NamespaceName: string(cenv.Namespace()),
777
PageSize: 10,
778
Query: visQuery,
779
})
780
ss.NoError(err)
781
if len(resp.Executions) != 1 {
782
return false
783
}
784
785
visRecord = resp.Executions[0]
786
return true
787
},
788
testcore.WaitForESToSettle,
789
100*time.Millisecond,
790
)
791
ss.Equal(storeID, visRecord.BusinessID)
792
793
// Verify TaskQueue is returned as a CHASM search attribute.
794
taskQueueVal, ok := chasm.SearchAttributeValue(visRecord.ChasmSearchAttributes, chasm.SearchAttributeTaskQueue)
795
ss.True(ok)
796
ss.Equal(tests.DefaultPayloadStoreTaskQueue, taskQueueVal)
797
})
798
}
799
800
func (s *ChasmSuite) TestMutableStateRebuilder() {
801
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
802
tv := testvars.New(ss.T())
803
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
804
defer cancel()
805
806
storeID := tv.Any().String()
807
_, err := tests.NewPayloadStoreHandler(
808
ctx,
809
tests.NewPayloadStoreRequest{
810
NamespaceID: cenv.NamespaceID(),
811
StoreID: storeID,
812
IDReusePolicy: chasm.BusinessIDReusePolicyRejectDuplicate,
813
IDConflictPolicy: chasm.BusinessIDConflictPolicyFail,
814
},
815
)
816
ss.NoError(err)
817
818
// Wait for the payload store to be visible.
819
visQuery := fmt.Sprintf("TemporalNamespaceDivision = '%d' AND WorkflowId = '%s'", tests.ArchetypeID, storeID)
820
var visRecord *chasm.VisibilityExecutionInfo[*testspb.TestPayloadStore]
821
var runID string
822
ss.AwaitTrue(
823
func() bool {
824
resp, err := chasm.ListExecutions[*tests.PayloadStore, *testspb.TestPayloadStore](ctx, &chasm.ListExecutionsRequest{
825
NamespaceName: string(cenv.Namespace()),
826
PageSize: 10,
827
Query: visQuery,
828
})
829
ss.NoError(err)
830
if len(resp.Executions) != 1 {
831
return false
832
}
833
834
visRecord = resp.Executions[0]
835
runID = visRecord.RunID
836
return true
837
},
838
testcore.WaitForESToSettle,
839
100*time.Millisecond,
840
)
841
ss.Equal(storeID, visRecord.BusinessID)
842
843
// payloadStore archetype is not the workflow archetype, should fail the rebuild.
844
ss.NotEqual(tests.Archetype, chasm.WorkflowArchetype, "Archetype should not be the workflow archetype")
845
846
_, err = cenv.AdminClient().RebuildMutableState(ss.Context(), &adminservice.RebuildMutableStateRequest{
847
Namespace: cenv.Namespace().String(),
848
Execution: &commonpb.WorkflowExecution{
849
WorkflowId: storeID,
850
RunId: runID,
851
},
852
})
853
ss.ErrorAs(err, new(*serviceerror.InvalidArgument))
854
})
855
}
856
857
func (s *ChasmSuite) TestUpdateWithStartExecution_UpdateExisting() {
858
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
859
tv := testvars.New(ss.T())
860
861
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
862
defer cancel()
863
864
storeID := tv.Any().String()
865
866
// Create initial PayloadStore.
867
createResp, err := tests.NewPayloadStoreHandler(
868
ctx,
869
tests.NewPayloadStoreRequest{
870
NamespaceID: cenv.NamespaceID(),
871
StoreID: storeID,
872
IDReusePolicy: chasm.BusinessIDReusePolicyAllowDuplicate,
873
IDConflictPolicy: chasm.BusinessIDConflictPolicyFail,
874
},
875
)
876
ss.NoError(err)
877
originalRunID := createResp.RunID
878
879
// Add a payload to the original store.
880
_, err = tests.AddPayloadHandler(
881
ctx,
882
tests.AddPayloadRequest{
883
NamespaceID: cenv.NamespaceID(),
884
StoreID: storeID,
885
PayloadKey: "original-key",
886
Payload: payload.EncodeString("original-value"),
887
},
888
)
889
ss.NoError(err)
890
891
// Call UpdateWithStartExecution - should update existing running execution.
892
newFnCalled := false
893
updateFnCalled := false
894
result, err := chasm.UpdateWithStartExecution(
895
ctx,
896
chasm.ExecutionKey{
897
NamespaceID: cenv.NamespaceID().String(),
898
BusinessID: storeID,
899
},
900
func(mutableContext chasm.MutableContext, _ any) (*tests.PayloadStore, error) {
901
newFnCalled = true
902
ss.Fail("newFn should not be called when execution exists and is running")
903
return nil, nil
904
},
905
func(store *tests.PayloadStore, mutableContext chasm.MutableContext, _ any) (any, error) {
906
updateFnCalled = true
907
// Update the store by closing it (marks for close but doesn't actually close yet).
908
store.State.Closed = true
909
return nil, nil
910
},
911
nil,
912
)
913
ss.NoError(err)
914
ss.False(newFnCalled, "newFn should not be called")
915
ss.True(updateFnCalled, "updateFn should be called")
916
ss.Equal(originalRunID, result.ExecutionKey.RunID, "RunID should be the same as original")
917
ss.NotNil(result.ExecutionRef)
918
919
// Verify the store was updated (closed flag set).
920
descResp, err := tests.DescribePayloadStoreHandler(
921
ctx,
922
tests.DescribePayloadStoreRequest{
923
NamespaceID: cenv.NamespaceID(),
924
StoreID: storeID,
925
},
926
)
927
ss.NoError(err)
928
ss.True(descResp.State.Closed, "Store should be marked as closed")
929
ss.Equal(int64(1), descResp.State.TotalCount) // Original payload still there.
930
})
931
}
932
933
func (s *ChasmSuite) TestUpdateWithStartExecution_CreateNew() {
934
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
935
tv := testvars.New(ss.T())
936
937
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
938
defer cancel()
939
940
storeID := tv.Any().String()
941
942
// Call UpdateWithStartExecution without creating execution first - should create new.
943
newFnCalled := false
944
updateFnCalled := false
945
result, err := chasm.UpdateWithStartExecution(
946
ctx,
947
chasm.ExecutionKey{
948
NamespaceID: cenv.NamespaceID().String(),
949
BusinessID: storeID,
950
},
951
func(mutableContext chasm.MutableContext, _ any) (*tests.PayloadStore, error) {
952
newFnCalled = true
953
store, err := tests.NewPayloadStore(mutableContext)
954
return store, err
955
},
956
func(store *tests.PayloadStore, mutableContext chasm.MutableContext, _ any) (any, error) {
957
updateFnCalled = true
958
// Apply update to the newly created store (like adding a signal during SignalWithStart).
959
store.State.TotalCount = 42
960
return nil, nil
961
},
962
nil,
963
)
964
ss.NoError(err)
965
ss.True(newFnCalled, "newFn should be called")
966
ss.True(updateFnCalled, "updateFn should be called after newFn")
967
ss.NotEmpty(result.ExecutionKey.RunID)
968
ss.NotNil(result.ExecutionRef)
969
970
// Verify the store was created with the update applied.
971
descResp, err := tests.DescribePayloadStoreHandler(
972
ctx,
973
tests.DescribePayloadStoreRequest{
974
NamespaceID: cenv.NamespaceID(),
975
StoreID: storeID,
976
},
977
)
978
ss.NoError(err)
979
ss.False(descResp.State.Closed, "Store should not be closed")
980
ss.Equal(int64(42), descResp.State.TotalCount) // Update was applied during creation.
981
})
982
}
983
984
func (s *ChasmSuite) TestPayloadStore_ApproximateExecutionSize() {
985
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
986
tv := testvars.New(ss.T())
987
988
ctx, cancel := context.WithTimeout(cenv.chasmCtx, chasmTestTimeout)
989
defer cancel()
990
991
storeID := tv.Any().String()
992
_, err := tests.NewPayloadStoreHandler(
993
ctx,
994
tests.NewPayloadStoreRequest{
995
NamespaceID: cenv.NamespaceID(),
996
StoreID: storeID,
997
},
998
)
999
ss.NoError(err)
1000
1001
descResp, err := tests.DescribePayloadStoreHandler(
1002
ctx,
1003
tests.DescribePayloadStoreRequest{
1004
NamespaceID: cenv.NamespaceID(),
1005
StoreID: storeID,
1006
},
1007
)
1008
ss.NoError(err)
1009
initialApproxSize := descResp.ApproximateStateSize
1010
1011
payloadSize := 100 * 1024 // 100KB
1012
payloadData := make([]byte, payloadSize)
1013
_, err = rand.Read(payloadData)
1014
ss.NoError(err)
1015
1016
_, err = tests.AddPayloadHandler(
1017
ctx,
1018
tests.AddPayloadRequest{
1019
NamespaceID: cenv.NamespaceID(),
1020
StoreID: storeID,
1021
PayloadKey: "key1",
1022
Payload: payload.EncodeBytes(payloadData),
1023
},
1024
)
1025
ss.NoError(err)
1026
1027
descResp, err = tests.DescribePayloadStoreHandler(
1028
ctx,
1029
tests.DescribePayloadStoreRequest{
1030
NamespaceID: cenv.NamespaceID(),
1031
StoreID: storeID,
1032
},
1033
)
1034
ss.NoError(err)
1035
currentApproxSize := descResp.ApproximateStateSize
1036
sizeDelta := float64(100) // Allow 100 bytes of variance due to overhead, encoding, etc.
1037
ss.InDelta(payloadSize, currentApproxSize-initialApproxSize, sizeDelta)
1038
1039
adminDescResp, err := cenv.AdminClient().DescribeMutableState(ss.Context(), &adminservice.DescribeMutableStateRequest{
1040
Namespace: cenv.Namespace().String(),
1041
Execution: &commonpb.WorkflowExecution{
1042
WorkflowId: storeID,
1043
},
1044
ArchetypeId: tests.ArchetypeID,
1045
})
1046
ss.NoError(err)
1047
ss.InDelta(adminDescResp.DatabaseMutableState.Size(), currentApproxSize, sizeDelta)
1048
})
1049
}
1050
1051
// TestNamespaceDelete_WithChasmExecutions verifies that running CHASM executions are cleaned
1052
// up when their namespace is deleted, exercising the DeleteExecution history service API.
1053
func (s *ChasmSuite) TestNamespaceDelete_WithChasmExecutions() {
1054
s.forBothConverters(func(ss *ChasmSuite, cenv chasmTestEnv) {
1055
tv := testvars.New(ss.T())
1056
1057
// Register a fresh namespace for this test.
1058
var namespaceSuffix [4]byte
1059
_, err := rand.Read(namespaceSuffix[:])
1060
ss.NoError(err)
1061
namespaceName := fmt.Sprintf("ns-chasm-delete-%x", namespaceSuffix)
1062
_, err = cenv.FrontendClient().RegisterNamespace(ss.Context(), &workflowservice.RegisterNamespaceRequest{
1063
Namespace: namespaceName,
1064
WorkflowExecutionRetentionPeriod: durationpb.New(24 * time.Hour),
1065
HistoryArchivalState: enumspb.ARCHIVAL_STATE_DISABLED,
1066
VisibilityArchivalState: enumspb.ARCHIVAL_STATE_DISABLED,
1067
})
1068
ss.NoError(err)
1069
1070
descResp, err := cenv.FrontendClient().DescribeNamespace(ss.Context(), &workflowservice.DescribeNamespaceRequest{
1071
Namespace: namespaceName,
1072
})
1073
ss.NoError(err)
1074
nsID := namespace.ID(descResp.GetNamespaceInfo().GetId())
1075
1076
// Create running CHASM executions in the new namespace.
1077
const numExecutions = 3
1078
for range numExecutions {
1079
_, err = tests.NewPayloadStoreHandler(cenv.chasmCtx, tests.NewPayloadStoreRequest{
1080
NamespaceID: nsID,
1081
StoreID: tv.Any().String(),
1082
IDReusePolicy: chasm.BusinessIDReusePolicyRejectDuplicate,
1083
IDConflictPolicy: chasm.BusinessIDConflictPolicyFail,
1084
})
1085
ss.NoError(err)
1086
}
1087
1088
// Wait for visibility records to appear.
1089
visQuery := fmt.Sprintf("TemporalNamespaceDivision = '%d'", tests.ArchetypeID)
1090
await.Require(ss.Context(), ss.T(), func(t *await.T) {
1091
resp, err := cenv.FrontendClient().ListWorkflowExecutions(t.Context(), &workflowservice.ListWorkflowExecutionsRequest{
1092
Namespace: namespaceName,
1093
PageSize: 10,
1094
Query: visQuery,
1095
})
1096
require.NoError(t, err)
1097
require.Len(t, resp.Executions, numExecutions)
1098
}, testcore.WaitForESToSettle, 100*time.Millisecond)
1099
1100
// Delete the namespace, which should trigger DeleteExecution for all CHASM executions.
1101
_, err = cenv.OperatorClient().DeleteNamespace(ss.Context(), &operatorservice.DeleteNamespaceRequest{
1102
Namespace: namespaceName,
1103
})
1104
ss.NoError(err)
1105
1106
// Verify all CHASM executions are cleaned up from visibility.
1107
await.Require(ss.Context(), ss.T(), func(t *await.T) {
1108
resp, err := cenv.FrontendClient().ListWorkflowExecutions(t.Context(), &workflowservice.ListWorkflowExecutionsRequest{
1109
Namespace: namespaceName,
1110
PageSize: 10,
1111
Query: visQuery,
1112
})
1113
var notFound *serviceerror.NamespaceNotFound
1114
if errors.As(err, ¬Found) {
1115
return // namespace fully deleted is also acceptable
1116
}
1117
require.NoError(t, err)
1118
require.Empty(t, resp.Executions)
1119
}, 20*time.Second*debug.TimeoutMultiplier, time.Second)
1120
})
1121
}
1122
1123
// TODO: More tests here...