go.temporal.io/server/tests/schedule_test.go
5393 LOC · 0 covered · 5393 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package tests
2
3
import (
4
"context"
5
"errors"
6
"fmt"
7
"net/http/httptest"
8
"strings"
9
"sync/atomic"
10
"testing"
11
"time"
12
13
"github.com/google/uuid"
14
"github.com/nexus-rpc/sdk-go/nexus"
15
"github.com/stretchr/testify/assert"
16
"github.com/stretchr/testify/require"
17
commonpb "go.temporal.io/api/common/v1"
18
enumspb "go.temporal.io/api/enums/v1"
19
historypb "go.temporal.io/api/history/v1"
20
schedulepb "go.temporal.io/api/schedule/v1"
21
"go.temporal.io/api/serviceerror"
22
taskqueuepb "go.temporal.io/api/taskqueue/v1"
23
workflowpb "go.temporal.io/api/workflow/v1"
24
"go.temporal.io/api/workflowservice/v1"
25
sdkclient "go.temporal.io/sdk/client"
26
"go.temporal.io/sdk/temporal"
27
"go.temporal.io/sdk/workflow"
28
schedulespb "go.temporal.io/server/api/schedule/v1"
29
"go.temporal.io/server/chasm"
30
"go.temporal.io/server/chasm/lib/callback"
31
chasmscheduler "go.temporal.io/server/chasm/lib/scheduler"
32
schedulerpb "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
33
"go.temporal.io/server/common/dynamicconfig"
34
"go.temporal.io/server/common/headers"
35
"go.temporal.io/server/common/nexus/nexusrpc"
36
"go.temporal.io/server/common/payload"
37
"go.temporal.io/server/common/payloads"
38
"go.temporal.io/server/common/primitives"
39
"go.temporal.io/server/common/searchattribute/sadefs"
40
"go.temporal.io/server/common/testing/await"
41
"go.temporal.io/server/common/testing/protorequire"
42
"go.temporal.io/server/service/worker/dummy"
43
"go.temporal.io/server/service/worker/scheduler"
44
"go.temporal.io/server/tests/testcore"
45
"google.golang.org/grpc/metadata"
46
"google.golang.org/protobuf/proto"
47
"google.golang.org/protobuf/types/known/durationpb"
48
"google.golang.org/protobuf/types/known/timestamppb"
49
)
50
51
// contextFactory wraps a base context for CHASM vs V1 differences.
52
type contextFactory func(context.Context) context.Context
53
54
var (
55
chasmContextFactory contextFactory = func(ctx context.Context) context.Context {
56
return metadata.NewOutgoingContext(ctx, metadata.Pairs(
57
headers.ExperimentHeaderName, "chasm-scheduler",
58
))
59
}
60
v1ContextFactory contextFactory = func(ctx context.Context) context.Context {
61
return ctx
62
}
63
)
64
65
func scheduleCommonOpts(t *testing.T) []testcore.TestOption {
66
opts := []testcore.TestOption{
67
testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true),
68
testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerSentinels, true),
69
testcore.WithDynamicConfig(dynamicconfig.FrontendAllowedExperiments, []string{"*"}),
70
}
71
if strings.HasPrefix(t.Name(), "TestScheduleV1") {
72
// only v1 needs the worker service
73
opts = append(opts, testcore.WithWorkerService("V1 scheduler"))
74
}
75
return opts
76
}
77
78
func newScheduleEnv(t *testing.T, opts ...testcore.TestOption) *testcore.TestEnv {
79
t.Helper()
80
opts = append(opts, testcore.WithDynamicConfig(dynamicconfig.FrontendAllowedExperiments, []string{"*"}))
81
env := testcore.NewEnv(t, opts...)
82
t.Cleanup(func() {
83
ctx, cancel := context.WithTimeout(chasmContextFactory(testcore.NewContext()), 30*time.Second)
84
defer cancel()
85
86
var scheduleIDs []string
87
var nextPageToken []byte
88
for {
89
response, err := env.FrontendClient().ListSchedules(ctx, &workflowservice.ListSchedulesRequest{
90
Namespace: env.Namespace().String(),
91
MaximumPageSize: 1000,
92
NextPageToken: nextPageToken,
93
})
94
if err != nil {
95
if t.Failed() {
96
t.Logf("schedule cleanup failed: list schedules: %v", err)
97
} else {
98
t.Errorf("schedule cleanup failed: list schedules: %v", err)
99
}
100
return
101
}
102
for _, schedule := range response.GetSchedules() {
103
scheduleIDs = append(scheduleIDs, schedule.GetScheduleId())
104
}
105
nextPageToken = response.GetNextPageToken()
106
if len(nextPageToken) == 0 {
107
break
108
}
109
}
110
111
var cleanupErr error
112
for _, scheduleID := range scheduleIDs {
113
_, err := env.FrontendClient().DeleteSchedule(ctx, &workflowservice.DeleteScheduleRequest{
114
Namespace: env.Namespace().String(),
115
ScheduleId: scheduleID,
116
Identity: "test cleanup",
117
})
118
var notFoundErr *serviceerror.NotFound
119
if err != nil && !errors.As(err, ¬FoundErr) {
120
cleanupErr = errors.Join(cleanupErr, fmt.Errorf("delete schedule %q: %w", scheduleID, err))
121
}
122
}
123
if cleanupErr != nil {
124
if t.Failed() {
125
t.Logf("schedule cleanup failed: %v", cleanupErr)
126
} else {
127
t.Errorf("schedule cleanup failed: %v", cleanupErr)
128
}
129
}
130
})
131
return env
132
}
133
134
const (
135
// fastInterval is the shortest interval that reliably ticks once per second.
136
fastInterval = 1 * time.Second
137
// noOpInterval is long enough that no spec tick fires during a test.
138
noOpInterval = 1 * time.Hour
139
// shortIdleTime is the CHASM IdleTime used by idle-close tests: short enough
140
// to keep tests fast, long enough to be reached reliably under load.
141
shortIdleTime = 3 * time.Second
142
// pollInterval is the poll cadence shared by all await/Never checks.
143
pollInterval = 200 * time.Millisecond
144
// awaitTimeout bounds how long an await waits for a condition to become true.
145
awaitTimeout = 30 * time.Second
146
// neverWindow is how long a Never check waits to confirm a condition stays false.
147
neverWindow = 5 * time.Second
148
)
149
150
// completeSignalName releases a workflow registered via registerGatedWorkflow.
151
const completeSignalName = "complete"
152
153
// intervalSpec is the single-interval schedule spec used by most tests.
154
func intervalSpec(every time.Duration) *schedulepb.ScheduleSpec {
155
return &schedulepb.ScheduleSpec{
156
Interval: []*schedulepb.IntervalSpec{{Interval: durationpb.New(every)}},
157
}
158
}
159
160
// newEnvWithIdleTime returns a schedule test env with the CHASM IdleTime
161
// tweakable set to idleTime, plus any extra options.
162
func newEnvWithIdleTime(t *testing.T, idleTime time.Duration, extra ...testcore.TestOption) *testcore.TestEnv {
163
tweakables := chasmscheduler.DefaultTweakables
164
tweakables.IdleTime = idleTime
165
opts := append(scheduleCommonOpts(t), testcore.WithDynamicConfig(chasmscheduler.CurrentTweakables, tweakables))
166
return newScheduleEnv(t, append(opts, extra...)...)
167
}
168
169
// createSchedule creates sched under sid and fails the test on error.
170
func createSchedule(ctx context.Context, t *testing.T, env *testcore.TestEnv, sid string, sched *schedulepb.Schedule) {
171
t.Helper()
172
_, err := env.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
173
Namespace: env.Namespace().String(),
174
ScheduleId: sid,
175
Schedule: sched,
176
Identity: "test",
177
RequestId: uuid.NewString(),
178
})
179
require.NoError(t, err)
180
}
181
182
// patchSchedule applies patch to sid and fails the test on error.
183
func patchSchedule(ctx context.Context, t *testing.T, env *testcore.TestEnv, sid string, patch *schedulepb.SchedulePatch) {
184
t.Helper()
185
_, err := env.FrontendClient().PatchSchedule(ctx, &workflowservice.PatchScheduleRequest{
186
Namespace: env.Namespace().String(),
187
ScheduleId: sid,
188
Patch: patch,
189
Identity: "test",
190
RequestId: uuid.NewString(),
191
})
192
require.NoError(t, err)
193
}
194
195
// backfillPatch builds a single-range backfill patch.
196
func backfillPatch(start, end time.Time, policy enumspb.ScheduleOverlapPolicy) *schedulepb.SchedulePatch {
197
return &schedulepb.SchedulePatch{
198
BackfillRequest: []*schedulepb.BackfillRequest{{
199
StartTime: timestamppb.New(start),
200
EndTime: timestamppb.New(end),
201
OverlapPolicy: policy,
202
}},
203
}
204
}
205
206
// triggerPatch builds a TriggerImmediately patch with the given overlap policy.
207
func triggerPatch(policy enumspb.ScheduleOverlapPolicy) *schedulepb.SchedulePatch {
208
return &schedulepb.SchedulePatch{
209
TriggerImmediately: &schedulepb.TriggerImmediatelyRequest{OverlapPolicy: policy},
210
}
211
}
212
213
// startWorkflowAction builds the StartWorkflow action shared by these tests.
214
func startWorkflowAction(env *testcore.TestEnv, wid, wt string) *schedulepb.ScheduleAction {
215
return &schedulepb.ScheduleAction{
216
Action: &schedulepb.ScheduleAction_StartWorkflow{
217
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
218
WorkflowId: wid,
219
WorkflowType: &commonpb.WorkflowType{Name: wt},
220
TaskQueue: &taskqueuepb.TaskQueue{Name: env.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
221
WorkflowIdReusePolicy: enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
222
},
223
},
224
}
225
}
226
227
// calendarSpec builds a single-instant calendar spec at the given time.
228
func calendarSpec(at time.Time) *schedulepb.CalendarSpec {
229
return &schedulepb.CalendarSpec{
230
Year: fmt.Sprintf("%d", at.Year()),
231
Month: at.Month().String(),
232
DayOfMonth: fmt.Sprintf("%d", at.Day()),
233
Hour: fmt.Sprintf("%d", at.Hour()),
234
Minute: fmt.Sprintf("%d", at.Minute()),
235
Second: fmt.Sprintf("%d", at.Second()),
236
}
237
}
238
239
// registerCountingWorkflow registers a workflow that records each execution in
240
// runs (via SideEffect, so replays don't double-count) and returns immediately.
241
//
242
// Each registered counting workflow should be associated with a distinct `runs`
243
// atomic.
244
func registerCountingWorkflow(env *testcore.TestEnv, wt string, runs *atomic.Int32) {
245
env.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
246
_ = workflow.SideEffect(ctx, func(workflow.Context) any { runs.Add(1); return 0 })
247
return nil
248
}, workflow.RegisterOptions{Name: wt})
249
}
250
251
// registerGatedWorkflow is like registerCountingWorkflow but the workflow stays
252
// running until the test signals completeSignalName (via completeRunningWorkflows).
253
func registerGatedWorkflow(env *testcore.TestEnv, wt string, runs *atomic.Int32) {
254
env.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
255
_ = workflow.SideEffect(ctx, func(workflow.Context) any { runs.Add(1); return 0 })
256
workflow.GetSignalChannel(ctx, completeSignalName).Receive(ctx, nil)
257
return nil
258
}, workflow.RegisterOptions{Name: wt})
259
}
260
261
// scheduleClosed reports whether the schedule has closed, i.e. DescribeSchedule
262
// returns NotFound specifically (not just any error).
263
func scheduleClosed(ctx context.Context, env *testcore.TestEnv, sid string) bool {
264
_, err := env.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
265
Namespace: env.Namespace().String(),
266
ScheduleId: sid,
267
})
268
var notFound *serviceerror.NotFound
269
return errors.As(err, ¬Found)
270
}
271
272
// completeRunningWorkflows signals completeSignalName to every running workflow
273
// of the schedule and returns the number it signaled.
274
func completeRunningWorkflows(ctx context.Context, t *testing.T, env *testcore.TestEnv, sid string) int {
275
t.Helper()
276
desc, err := env.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
277
Namespace: env.Namespace().String(),
278
ScheduleId: sid,
279
})
280
require.NoError(t, err)
281
running := desc.GetInfo().GetRunningWorkflows()
282
for _, wf := range running {
283
_, err := env.FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
284
Namespace: env.Namespace().String(),
285
WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: wf.GetWorkflowId()},
286
SignalName: completeSignalName,
287
Identity: "test",
288
RequestId: uuid.NewString(),
289
})
290
require.NoError(t, err)
291
}
292
return len(running)
293
}
294
295
// terminalStop selects how a fired run is stopped in
296
// testPauseOnFailureIgnoresCancelTerminate.
297
type terminalStop int
298
299
const (
300
stopByCancel terminalStop = iota
301
stopByTerminate
302
)
303
304
// testPauseOnFailureIgnoresCancelTerminate verifies that manually canceling or
305
// terminating a fired run of a PauseOnFailure schedule does NOT pause the
306
// schedule. CanceledTerminatedCountAsFailures defaults to false, so a cancel or
307
// terminate is a routine operation, not an application failure -- matching V1.
308
//
309
// Regression guard for the CHASM HandleNexusCompletion fix: previously any
310
// non-COMPLETED status paused a PauseOnFailure schedule (and terminated runs
311
// were mislabeled FAILED), so a manual cancel/terminate would silently stop all
312
// future runs.
313
func testPauseOnFailureIgnoresCancelTerminate(t *testing.T, newContext contextFactory, stop terminalStop) {
314
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
315
ctx := newContext(testcore.NewContext())
316
317
sid := testcore.RandomizeStr("sched-pauseonfail")
318
wid := testcore.RandomizeStr("sched-pauseonfail-wf")
319
wt := testcore.RandomizeStr("sched-pauseonfail-wt")
320
321
// A run that stays running until it is canceled or terminated. On
322
// cancellation workflow.Sleep returns the cancellation error, which the
323
// workflow returns so the run closes as CANCELED (rather than COMPLETED); a
324
// terminate closes it forcefully as TERMINATED.
325
s.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
326
return workflow.Sleep(ctx, time.Hour)
327
}, workflow.RegisterOptions{Name: wt})
328
329
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
330
Spec: intervalSpec(fastInterval),
331
Action: startWorkflowAction(s, wid, wt),
332
Policies: &schedulepb.SchedulePolicies{PauseOnFailure: true},
333
})
334
335
describe := func() *workflowservice.DescribeScheduleResponse {
336
d, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
337
Namespace: s.Namespace().String(),
338
ScheduleId: sid,
339
})
340
require.NoError(t, err)
341
return d
342
}
343
344
// Wait for the schedule to fire a run that is RUNNING, and capture its actual
345
// started execution (the started workflow id may differ from the configured
346
// wid, e.g. a timestamp suffix, so read it from the action result rather than
347
// assuming wid).
348
var exec *commonpb.WorkflowExecution
349
require.Eventually(t, func() bool {
350
for _, a := range describe().GetInfo().GetRecentActions() {
351
if a.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING {
352
exec = a.GetStartWorkflowResult()
353
return exec.GetRunId() != ""
354
}
355
}
356
return false
357
}, awaitTimeout, pollInterval, "schedule should fire a running workflow")
358
runID := exec.GetRunId()
359
360
// Stop that run the way a user would.
361
wantStatus := enumspb.WORKFLOW_EXECUTION_STATUS_CANCELED
362
if stop == stopByTerminate {
363
wantStatus = enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED
364
_, err := s.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
365
Namespace: s.Namespace().String(),
366
WorkflowExecution: exec,
367
Reason: t.Name(),
368
Identity: "test",
369
})
370
require.NoError(t, err)
371
} else {
372
_, err := s.FrontendClient().RequestCancelWorkflowExecution(ctx, &workflowservice.RequestCancelWorkflowExecutionRequest{
373
Namespace: s.Namespace().String(),
374
WorkflowExecution: exec,
375
Identity: "test",
376
RequestId: uuid.NewString(),
377
})
378
require.NoError(t, err)
379
}
380
381
// Once the run's terminal status is recorded on the schedule,
382
// HandleNexusCompletion has processed the completion -- and, in the same
383
// mutation, made (or correctly declined) the pause-on-failure decision. So
384
// reading Paused from the SAME describe response as the terminal status is
385
// race-free: no later pause can slip in.
386
var pausedAfter bool
387
require.Eventually(t, func() bool {
388
d := describe()
389
for _, a := range d.GetInfo().GetRecentActions() {
390
if a.GetStartWorkflowResult().GetRunId() == runID && a.GetStartWorkflowStatus() == wantStatus {
391
pausedAfter = d.GetSchedule().GetState().GetPaused()
392
return true
393
}
394
}
395
return false
396
}, awaitTimeout, pollInterval, "run %s should reach %s and be recorded on the schedule", runID, wantStatus)
397
398
require.False(t, pausedAfter,
399
"a %s workflow must not pause a PauseOnFailure schedule by default "+
400
"(CanceledTerminatedCountAsFailures defaults to false)", wantStatus)
401
}
402
403
func TestScheduleCHASM(t *testing.T) {
404
t.Parallel()
405
runSharedScheduleTests(t, chasmContextFactory)
406
407
// CHASM-only tests
408
newContext := chasmContextFactory
409
t.Run("TestCreateScheduleAlreadyExists", func(t *testing.T) { t.Parallel(); testCreateScheduleAlreadyExists(t, newContext) })
410
t.Run("TestCreateScheduleDuplicateSdkError", func(t *testing.T) { t.Parallel(); testCreateScheduleDuplicateSdkError(t, true) })
411
t.Run("TestPatchRejectsExcessBackfillers", func(t *testing.T) { t.Parallel(); testPatchRejectsExcessBackfillers(t, newContext) })
412
t.Run("TestDoubleReset_HSMCallbacks", func(t *testing.T) { t.Parallel(); testScheduledWorkflowDoubleReset(t, newContext, false) })
413
t.Run("TestDoubleReset_ChasmCallbacks", func(t *testing.T) { t.Parallel(); testScheduledWorkflowDoubleReset(t, newContext, true) })
414
t.Run("TestResetWithAdditionalCallback_HSMCallbacks", func(t *testing.T) { t.Parallel(); testResetWithAdditionalCallback(t, newContext, false) })
415
t.Run("TestResetWithAdditionalCallback_ChasmCallbacks", func(t *testing.T) { t.Parallel(); testResetWithAdditionalCallback(t, newContext, true) })
416
t.Run("TestMigrationCallbackAttach", func(t *testing.T) { t.Parallel(); testMigrationCallbackAttach(t, newContext) })
417
t.Run("TestCreatesWorkflowSentinel", func(t *testing.T) { t.Parallel(); testCreatesWorkflowSentinel(t, newContext) })
418
t.Run("TestSkipsWorkflowSentinelWhenDisabled", func(t *testing.T) { t.Parallel(); testSkipsWorkflowSentinelWhenDisabled(t, newContext) })
419
t.Run("TestUpdateScheduleMemo", func(t *testing.T) { t.Parallel(); testUpdateScheduleMemo(t, newContext) })
420
t.Run("TestUpdateScheduleMemoOnly", func(t *testing.T) { t.Parallel(); testUpdateScheduleMemoOnly(t, newContext) })
421
t.Run("TestStateSizeBytesReported", func(t *testing.T) { t.Parallel(); testStateSizeBytesReported(t, newContext) })
422
t.Run("TestBufferOverrunDropsActions", func(t *testing.T) { t.Parallel(); testBufferOverrunDropsActions(t, newContext) })
423
t.Run("IdleClose", func(t *testing.T) {
424
t.Parallel()
425
testScheduleClosesFromIdle(t, newContext)
426
t.Run("ManualOnly", func(t *testing.T) { t.Parallel(); testManualOnlyUnpausedClosesFromIdle(t, newContext) })
427
t.Run("PauseDuringWindow", func(t *testing.T) { t.Parallel(); testPauseDuringIdleWindow(t, newContext) })
428
t.Run("BackfillBlocks", func(t *testing.T) { t.Parallel(); testBackfillBlocksIdleClose(t, newContext) })
429
})
430
t.Run("PausedBehavior", func(t *testing.T) {
431
t.Parallel()
432
t.Run("DropsCatchup", func(t *testing.T) { t.Parallel(); testPausedDropsCatchup(t, newContext) })
433
t.Run("RecentActionsAdvance", func(t *testing.T) { t.Parallel(); testRecentActionsAdvanceWhilePaused(t, newContext) })
434
t.Run("FutureActionTimesAdvance", func(t *testing.T) { t.Parallel(); testFutureActionTimesAdvanceWhilePaused(t, newContext) })
435
t.Run("BackfillDrains", func(t *testing.T) { t.Parallel(); testBackfillOnPausedSchedule(t, newContext) })
436
})
437
t.Run("TestScheduledWorkflowContinueAsNewCompletion", func(t *testing.T) { t.Parallel(); testScheduledWorkflowContinueAsNewCompletion(t, newContext) })
438
t.Run("PauseOnFailure_CancelDoesNotPause", func(t *testing.T) {
439
t.Parallel()
440
testPauseOnFailureIgnoresCancelTerminate(t, newContext, stopByCancel)
441
})
442
t.Run("PauseOnFailure_TerminateDoesNotPause", func(t *testing.T) {
443
t.Parallel()
444
testPauseOnFailureIgnoresCancelTerminate(t, newContext, stopByTerminate)
445
})
446
}
447
448
func TestScheduleV1(t *testing.T) {
449
t.Parallel()
450
runSharedScheduleTests(t, v1ContextFactory)
451
452
// V1-only tests
453
newContext := v1ContextFactory
454
t.Run("TestCreateScheduleDuplicateSdkError", func(t *testing.T) { t.Parallel(); testCreateScheduleDuplicateSdkError(t, false) })
455
t.Run("TestCHASMCanListV1Schedules", func(t *testing.T) { t.Parallel(); testCHASMCanListV1Schedules(t, newContext) })
456
t.Run("TestRefresh", func(t *testing.T) { t.Parallel(); testRefresh(t, newContext) })
457
t.Run("TestListBeforeRun", func(t *testing.T) { t.Parallel(); testListBeforeRun(t, newContext) })
458
t.Run("TestRateLimit", func(t *testing.T) { t.Parallel(); testRateLimit(t, newContext) })
459
t.Run("TestNextTimeCache", func(t *testing.T) { t.Parallel(); testNextTimeCache(t, newContext) })
460
t.Run("TestCreatesCHASMSentinel", func(t *testing.T) { t.Parallel(); testCreatesCHASMSentinel(t, newContext) })
461
t.Run("TestSkipsCHASMSentinelWhenDisabled", func(t *testing.T) { t.Parallel(); testSkipsCHASMSentinelWhenDisabled(t, newContext) })
462
t.Run("TestUpdateScheduleMemoRejected", func(t *testing.T) { t.Parallel(); testUpdateScheduleMemoRejected(t, newContext) })
463
}
464
465
func runSharedScheduleTests(t *testing.T, newContext contextFactory) {
466
t.Run("TestBasics", func(t *testing.T) { t.Parallel(); testBasics(t, newContext) })
467
t.Run("TestInput", func(t *testing.T) { t.Parallel(); testInput(t, newContext) })
468
t.Run("TestLastCompletionAndError", func(t *testing.T) { t.Parallel(); testLastCompletionAndError(t, newContext) })
469
t.Run("TestScheduleContinuesAfterWorkflowRetryFailure", func(t *testing.T) { t.Parallel(); testScheduleContinuesAfterWorkflowRetryFailure(t, newContext) })
470
t.Run("TestListSchedulesReturnsWorkflowStatus", func(t *testing.T) { t.Parallel(); testListSchedulesReturnsWorkflowStatus(t, newContext) })
471
t.Run("TestUpdateIntervalTakesEffect", func(t *testing.T) { t.Parallel(); testUpdateIntervalTakesEffect(t, newContext) })
472
t.Run("TestListScheduleMatchingTimes", func(t *testing.T) { t.Parallel(); testListScheduleMatchingTimes(t, newContext) })
473
t.Run("TestLimitMemoSpecSize", func(t *testing.T) { t.Parallel(); testLimitMemoSpecSize(t, newContext) })
474
t.Run("TestCountSchedules", func(t *testing.T) { t.Parallel(); testCountSchedules(t, newContext) })
475
t.Run("TestSchedule_InternalTaskQueue", func(t *testing.T) { t.Parallel(); testScheduleInternalTaskQueue(t, newContext) })
476
t.Run("TestDeletedScheduleOperations", func(t *testing.T) { t.Parallel(); testDeletedScheduleOperations(t, newContext) })
477
t.Run("TestUnpauseResumesProcessing", func(t *testing.T) { t.Parallel(); testCHASMUnpauseResumesProcessing(t, newContext) })
478
t.Run("TestPausedScheduleNeverIdles", func(t *testing.T) { t.Parallel(); testPausedScheduleNeverIdles(t, newContext) })
479
t.Run("TestPausedEmptySpecStaysOpen", func(t *testing.T) { t.Parallel(); testPausedEmptySpecStaysOpen(t, newContext) })
480
t.Run("TriggerImmediately", func(t *testing.T) {
481
t.Parallel()
482
t.Run("OnActiveSchedule", func(t *testing.T) { t.Parallel(); testTriggerImmediatelyOnActiveSchedule(t, newContext) })
483
t.Run("OnPausedSchedule", func(t *testing.T) { t.Parallel(); testTriggerImmediatelyOnPausedSchedule(t, newContext) })
484
t.Run("AfterActionsExhausted", func(t *testing.T) { t.Parallel(); testTriggerImmediatelyAfterActionsExhausted(t, newContext) })
485
})
486
t.Run("Backfill", func(t *testing.T) {
487
t.Parallel()
488
t.Run("ReprocessCompletedActionExactTimePaused", func(t *testing.T) {
489
t.Parallel()
490
testBackfillReprocessesCompletedAction(t, newContext, true, 0)
491
})
492
t.Run("ReprocessCompletedActionExactTimeActive", func(t *testing.T) {
493
t.Parallel()
494
testBackfillReprocessesCompletedAction(t, newContext, false, 0)
495
})
496
t.Run("ReprocessCompletedActionInRange", func(t *testing.T) {
497
t.Parallel()
498
testBackfillReprocessesCompletedAction(t, newContext, true, 1)
499
})
500
t.Run("SkipOverlap", func(t *testing.T) { t.Parallel(); testBackfillWithSkipOverlap(t, newContext) })
501
t.Run("BufferOneOverlap", func(t *testing.T) { t.Parallel(); testBackfillWithBufferOneOverlap(t, newContext) })
502
t.Run("MultiRangeCountedExactlyOnce", func(t *testing.T) { t.Parallel(); testMultiRangeBackfillCountedExactlyOnce(t, newContext) })
503
t.Run("RangeSmallerThanInterval", func(t *testing.T) { t.Parallel(); testBackfillRangeSmallerThanInterval(t, newContext) })
504
})
505
t.Run("TestUpdateScheduleRequestIDTooLong", func(t *testing.T) { t.Parallel(); testUpdateScheduleRequestIDTooLong(t, newContext) })
506
t.Run("TestUpdateScheduleBlobSizeLimit", func(t *testing.T) { t.Parallel(); testUpdateScheduleBlobSizeLimit(t, newContext) })
507
t.Run("TestListSchedulesPagination", func(t *testing.T) { t.Parallel(); testListSchedulesPagination(t, newContext) })
508
t.Run("TestListSchedulesFilterAndEntryFields", func(t *testing.T) { t.Parallel(); testListSchedulesFilterAndEntryFields(t, newContext) })
509
t.Run("TestListSchedulesFilterByScheduleId", func(t *testing.T) { t.Parallel(); testListSchedulesFilterByScheduleID(t, newContext) })
510
t.Run("TestBufferSizeReportedWhenBuffered", func(t *testing.T) { t.Parallel(); testBufferSizeReportedWhenBuffered(t, newContext) })
511
t.Run("TestBufferOneDeferredFiresAfterCompletion", func(t *testing.T) { t.Parallel(); testBufferOneDeferredFiresAfterCompletion(t, newContext) })
512
}
513
514
// testBufferSizeReportedWhenBuffered verifies that ScheduleInfo.BufferSize is
515
// populated by both the V1 and V2 (CHASM) schedulers when at least one fire is
516
// queued behind a still-running workflow. A schedule with a 1s interval and
517
// BUFFER_ONE keeps exactly one start in the buffer while the first workflow
518
// is still running.
519
func testBufferSizeReportedWhenBuffered(t *testing.T, newContext contextFactory) {
520
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
521
522
sid := testcore.RandomizeStr("sched-buffer-size")
523
wid := testcore.RandomizeStr("sched-buffer-size-wf")
524
wt := testcore.RandomizeStr("sched-buffer-size-wt")
525
526
s.SdkWorker().RegisterWorkflowWithOptions(
527
func(ctx workflow.Context) error {
528
return workflow.Sleep(ctx, time.Hour)
529
},
530
workflow.RegisterOptions{Name: wt},
531
)
532
533
schedule := &schedulepb.Schedule{
534
Spec: &schedulepb.ScheduleSpec{
535
Interval: []*schedulepb.IntervalSpec{
536
{Interval: durationpb.New(1 * time.Second)},
537
},
538
},
539
Action: &schedulepb.ScheduleAction{
540
Action: &schedulepb.ScheduleAction_StartWorkflow{
541
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
542
WorkflowId: wid,
543
WorkflowType: &commonpb.WorkflowType{Name: wt},
544
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
545
},
546
},
547
},
548
Policies: &schedulepb.SchedulePolicies{
549
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE,
550
},
551
}
552
553
ctx := newContext(s.Context())
554
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
555
Namespace: s.Namespace().String(),
556
ScheduleId: sid,
557
Schedule: schedule,
558
Identity: "test",
559
RequestId: uuid.NewString(),
560
})
561
s.NoError(err)
562
563
var lastDescribe *workflowservice.DescribeScheduleResponse
564
s.Eventually(func() bool {
565
desc, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
566
Namespace: s.Namespace().String(),
567
ScheduleId: sid,
568
})
569
if descErr != nil {
570
return false
571
}
572
lastDescribe = desc
573
return desc.GetInfo().GetBufferSize() >= 1 && len(desc.GetInfo().GetRunningWorkflows()) >= 1
574
}, 30*time.Second, 500*time.Millisecond, "DescribeSchedule should report BufferSize >= 1 with a running workflow blocking the buffer")
575
576
s.GreaterOrEqual(lastDescribe.GetInfo().GetBufferSize(), int64(1), "BufferSize must reflect at least one buffered start")
577
s.GreaterOrEqual(len(lastDescribe.GetInfo().GetRunningWorkflows()), 1, "expected the buffered fire to be queued behind a running workflow")
578
}
579
580
// A full buffer (MaxBufferSize reached) makes the CHASM generator drop further
581
// fires and increment ScheduleInfo.BufferDropped.
582
func testBufferOverrunDropsActions(t *testing.T, newContext contextFactory) {
583
// A small buffer plus a gated workflow under BUFFER_ALL fills within a few
584
// fast-interval ticks. Only the CHASM scheduler reads these tweakables.
585
tweakables := chasmscheduler.DefaultTweakables
586
tweakables.MaxBufferSize = 2
587
opts := append(scheduleCommonOpts(t), testcore.WithDynamicConfig(chasmscheduler.CurrentTweakables, tweakables))
588
s := testcore.NewEnv(t, opts...)
589
590
sid := testcore.RandomizeStr("sched-buffer-overrun")
591
wid := testcore.RandomizeStr("sched-buffer-overrun-wf")
592
wt := testcore.RandomizeStr("sched-buffer-overrun-wt")
593
594
var runs atomic.Int32
595
registerGatedWorkflow(s, wt, &runs)
596
597
ctx := newContext(s.Context()) //nolint:staticcheck // SA1019
598
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
599
Spec: intervalSpec(fastInterval),
600
Action: startWorkflowAction(s, wid, wt),
601
Policies: &schedulepb.SchedulePolicies{
602
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
603
},
604
})
605
606
// RunningWorkflows lags the drop (it needs the async StartWorkflow to land), so
607
// assert both together rather than after the wait.
608
require.Eventually(t, func() bool {
609
desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
610
Namespace: s.Namespace().String(),
611
ScheduleId: sid,
612
})
613
if err != nil {
614
return false
615
}
616
return desc.GetInfo().GetBufferDropped() > 0 &&
617
len(desc.GetInfo().GetRunningWorkflows()) >= 1
618
}, awaitTimeout, pollInterval, "expected BufferDropped > 0 with a gated workflow holding the buffer full")
619
620
require.Positive(t, runs.Load(), "the gated workflow should have started")
621
622
completeRunningWorkflows(ctx, t, s, sid)
623
}
624
625
// testRecentActionsAdvanceWhilePaused verifies that an in-flight workflow's
626
// completion status surfaces in ListSchedules' RecentActions even while the
627
// schedule is paused. CHASM-only: the live Invoker updates BufferedStart on
628
// Nexus completion regardless of paused state, so the listed status moves
629
// from RUNNING to COMPLETED. V1's scheduler workflow does not advance its
630
// memo while paused, so the listed status stays at RUNNING until unpause.
631
func testRecentActionsAdvanceWhilePaused(t *testing.T, newContext contextFactory) {
632
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
633
634
sid := testcore.RandomizeStr("sched-recentactions-paused")
635
wid := testcore.RandomizeStr("sched-recentactions-paused-wf")
636
wt := testcore.RandomizeStr("sched-recentactions-paused-wt")
637
638
// Gate the run so the RUNNING -> COMPLETED transition is driven explicitly.
639
var runs atomic.Int32
640
registerGatedWorkflow(s, wt, &runs)
641
642
ctx := newContext(s.Context())
643
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
644
Spec: intervalSpec(fastInterval),
645
Action: startWorkflowAction(s, wid, wt),
646
})
647
648
// Wait for the first workflow to be reported as RUNNING in ListSchedules.
649
running := getScheduleEntryFromVisibility(s, sid, newContext, func(ent *schedulepb.ScheduleListEntry) bool {
650
return len(ent.Info.RecentActions) >= 1 &&
651
ent.Info.RecentActions[0].GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING
652
})
653
runningRunID := running.Info.RecentActions[0].GetStartWorkflowResult().GetRunId()
654
require.NotEmpty(t, runningRunID)
655
656
// Pause, then release the run: its COMPLETED status must surface while paused.
657
patchSchedule(ctx, t, s, sid, &schedulepb.SchedulePatch{Pause: "pausing for the test"})
658
659
signaled := completeRunningWorkflows(ctx, t, s, sid)
660
require.Equal(t, 1, signaled, "exactly one run should be in flight under the default SKIP overlap policy")
661
662
// While paused, the listed RecentActions entry transitions to COMPLETED.
663
getScheduleEntryFromVisibility(s, sid, newContext, func(ent *schedulepb.ScheduleListEntry) bool {
664
for _, a := range ent.Info.RecentActions {
665
if a.GetStartWorkflowResult().GetRunId() == runningRunID &&
666
a.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED {
667
return true
668
}
669
}
670
return false
671
})
672
require.Equal(t, int32(1), runs.Load(), "exactly one run should have fired (SKIP overlap, then paused)")
673
}
674
675
// testFutureActionTimesAdvanceWhilePaused verifies that ListSchedules returns
676
// up-to-date FutureActionTimes for a paused schedule. CHASM's always-on
677
// Generator advances the high water mark and rebuilds FutureActionTimes
678
// against the spec even when paused, so listed times stay in the future and
679
// never roll into the past. The legacy V1 scheduler workflow does not advance
680
// while paused, so its projected times would freeze at pause time and
681
// eventually all sit in the past - hence this test is registered CHASM-only.
682
func testFutureActionTimesAdvanceWhilePaused(t *testing.T, newContext contextFactory) {
683
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
684
685
sid := testcore.RandomizeStr("sched-future-actions-paused")
686
wid := testcore.RandomizeStr("sched-future-actions-paused-wf")
687
wt := testcore.RandomizeStr("sched-future-actions-paused-wt")
688
689
var runs atomic.Int32
690
registerCountingWorkflow(s, wt, &runs)
691
692
ctx := newContext(s.Context())
693
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
694
Spec: intervalSpec(fastInterval),
695
State: &schedulepb.ScheduleState{Paused: true},
696
Action: startWorkflowAction(s, wid, wt),
697
})
698
699
// Wait for visibility to surface an initial FutureActionTimes projection.
700
initial := getScheduleEntryFromVisibility(s, sid, newContext, func(ent *schedulepb.ScheduleListEntry) bool {
701
return len(ent.Info.FutureActionTimes) > 0
702
})
703
initialFirst := initial.Info.FutureActionTimes[0].AsTime()
704
705
// While still paused, the earliest projected time must advance past the
706
// initial value: the Generator keeps ticking and republishing the
707
// projection, even though no workflows fire.
708
getScheduleEntryFromVisibility(s, sid, newContext, func(ent *schedulepb.ScheduleListEntry) bool {
709
return len(ent.Info.FutureActionTimes) > 0 &&
710
ent.Info.FutureActionTimes[0].AsTime().After(initialFirst)
711
})
712
require.Zero(t, runs.Load(), "a paused schedule must not fire any workflows")
713
}
714
715
// testBufferOneDeferredFiresAfterCompletion exercises the BUFFER_ONE deferred
716
// lifecycle end-to-end: an action that gets buffered while a workflow is
717
// running must fire once that workflow completes. Without re-enabling the
718
// deferred start (Attempt=-1 -> 0 in recordCompletedAction), the buffered
719
// fire would be stranded.
720
func testBufferOneDeferredFiresAfterCompletion(t *testing.T, newContext contextFactory) {
721
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
722
723
sid := testcore.RandomizeStr("sched-buffer-one-deferred")
724
wid := testcore.RandomizeStr("sched-buffer-one-deferred-wf")
725
wt := testcore.RandomizeStr("sched-buffer-one-deferred-wt")
726
727
// Gate runs so "first running, second buffered" and "deferred fires after
728
// completion" are both reached deterministically.
729
var runs atomic.Int32
730
registerGatedWorkflow(s, wt, &runs)
731
732
ctx := newContext(s.Context())
733
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
734
Spec: intervalSpec(fastInterval),
735
Action: startWorkflowAction(s, wid, wt),
736
Policies: &schedulepb.SchedulePolicies{
737
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE,
738
},
739
})
740
741
// Exactly one workflow runs with exactly one start buffered behind it (BUFFER_ONE caps the buffer at one).
742
await.RequireTruef(t, func() bool {
743
desc, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
744
Namespace: s.Namespace().String(),
745
ScheduleId: sid,
746
})
747
return descErr == nil && desc.GetInfo().GetBufferSize() == 1 && len(desc.GetInfo().GetRunningWorkflows()) == 1
748
}, awaitTimeout, pollInterval, "expected exactly one running workflow with one deferred start buffered behind it")
749
require.Equal(t, int32(1), runs.Load(), "only the first workflow should have fired before the running one completes")
750
751
// Releasing the running workflow must re-enable the deferred start (Attempt=-1 -> 0) so it fires.
752
require.Equal(t, 1, completeRunningWorkflows(ctx, t, s, sid))
753
await.RequireTruef(t, func() bool { return runs.Load() == 2 },
754
awaitTimeout, pollInterval,
755
"deferred start must fire after the running workflow completes - regression for the Attempt=-1 -> 0 re-enable path")
756
757
// The fire is specifically the tick buffered directly behind the first start,
758
// not a fresh action generated after completion. RecentActions lists the
759
// completed first tick and the still-running deferred start; with no jitter the
760
// deferred start's nominal time is exactly one interval after the first tick's.
761
await.RequireTruef(t, func() bool {
762
desc, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
763
Namespace: s.Namespace().String(),
764
ScheduleId: sid,
765
})
766
if descErr != nil {
767
return false
768
}
769
var first, deferred *schedulepb.ScheduleActionResult
770
for _, r := range desc.GetInfo().GetRecentActions() {
771
if r.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING {
772
deferred = r
773
} else {
774
first = r
775
}
776
}
777
return first != nil && deferred != nil &&
778
deferred.GetScheduleTime().AsTime().Sub(first.GetScheduleTime().AsTime()) == fastInterval
779
}, awaitTimeout, pollInterval,
780
"deferred fire must be the start buffered one interval after the first tick, not a later fresh action")
781
}
782
783
func testDeletedScheduleOperations(t *testing.T, newContext contextFactory) {
784
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
785
786
sid := "sched-test-deleted-ops"
787
wid := "sched-test-deleted-ops-wf"
788
wt := "sched-test-deleted-ops-wt"
789
790
schedule := &schedulepb.Schedule{
791
Spec: &schedulepb.ScheduleSpec{
792
Interval: []*schedulepb.IntervalSpec{
793
{Interval: durationpb.New(1 * time.Hour)},
794
},
795
},
796
Action: &schedulepb.ScheduleAction{
797
Action: &schedulepb.ScheduleAction_StartWorkflow{
798
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
799
WorkflowId: wid,
800
WorkflowType: &commonpb.WorkflowType{Name: wt},
801
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
802
},
803
},
804
},
805
}
806
807
// Create a schedule.
808
_, err := s.FrontendClient().CreateSchedule(newContext(s.Context()), &workflowservice.CreateScheduleRequest{
809
Namespace: s.Namespace().String(),
810
ScheduleId: sid,
811
Schedule: schedule,
812
Identity: "test",
813
RequestId: uuid.NewString(),
814
})
815
s.NoError(err)
816
817
// Delete the schedule.
818
_, err = s.FrontendClient().DeleteSchedule(newContext(s.Context()), &workflowservice.DeleteScheduleRequest{
819
Namespace: s.Namespace().String(),
820
ScheduleId: sid,
821
Identity: "test",
822
})
823
s.NoError(err)
824
825
// Describe should return NotFound.
826
var notFoundErr *serviceerror.NotFound
827
s.Eventually(func() bool {
828
_, descErr := s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
829
Namespace: s.Namespace().String(),
830
ScheduleId: sid,
831
})
832
return errors.As(descErr, ¬FoundErr)
833
}, 10*time.Second, 200*time.Millisecond)
834
835
// Update, Patch, and Delete behave differently across CHASM and V1,
836
// so they are not tested here. See TestScheduleUpdateAfterDelete.
837
}
838
839
func testBasics(t *testing.T, newContext contextFactory) {
840
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
841
842
sid := "sched-test-basics"
843
wid := "sched-test-basics-wf"
844
wt := "sched-test-basics-wt"
845
wt2 := "sched-test-basics-wt2"
846
// switch this to test with search attribute mapper:
847
// csaKeyword := "AliasForCustomKeywordField"
848
csaKeyword := "CustomKeywordField"
849
csaInt := "CustomIntField"
850
csaBool := "CustomBoolField"
851
852
wfMemo := payload.EncodeString("workflow memo")
853
wfSAValue := payload.EncodeString("workflow sa value")
854
schMemo := payload.EncodeString("schedule memo")
855
schSAValue := payload.EncodeString("schedule sa value")
856
schSAIntValue, _ := payload.Encode(123)
857
schSABoolValue, _ := payload.Encode(true)
858
859
schedule := &schedulepb.Schedule{
860
Spec: &schedulepb.ScheduleSpec{
861
Interval: []*schedulepb.IntervalSpec{
862
{Interval: durationpb.New(5 * time.Second)},
863
},
864
Calendar: []*schedulepb.CalendarSpec{
865
{DayOfMonth: "10", Year: "2010"},
866
},
867
CronString: []string{"11 11/11 11 11 1 2011"},
868
},
869
Action: &schedulepb.ScheduleAction{
870
Action: &schedulepb.ScheduleAction_StartWorkflow{
871
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
872
WorkflowId: wid,
873
WorkflowType: &commonpb.WorkflowType{Name: wt},
874
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
875
Memo: &commonpb.Memo{
876
Fields: map[string]*commonpb.Payload{"wfmemo1": wfMemo},
877
},
878
SearchAttributes: &commonpb.SearchAttributes{
879
IndexedFields: map[string]*commonpb.Payload{csaKeyword: wfSAValue},
880
},
881
},
882
},
883
},
884
}
885
req := &workflowservice.CreateScheduleRequest{
886
Namespace: s.Namespace().String(),
887
ScheduleId: sid,
888
Schedule: schedule,
889
Identity: "test",
890
RequestId: uuid.NewString(),
891
Memo: &commonpb.Memo{
892
Fields: map[string]*commonpb.Payload{"schedmemo1": schMemo},
893
},
894
SearchAttributes: &commonpb.SearchAttributes{
895
IndexedFields: map[string]*commonpb.Payload{
896
csaKeyword: schSAValue,
897
csaInt: schSAIntValue,
898
csaBool: schSABoolValue,
899
},
900
},
901
}
902
903
var runs, runs2 int32
904
workflowFn := func(ctx workflow.Context) error {
905
workflow.SideEffect(ctx, func(ctx workflow.Context) any {
906
atomic.AddInt32(&runs, 1)
907
return 0
908
})
909
return nil
910
}
911
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
912
workflow2Fn := func(ctx workflow.Context) error {
913
workflow.SideEffect(ctx, func(ctx workflow.Context) any {
914
atomic.AddInt32(&runs2, 1)
915
return 0
916
})
917
return nil
918
}
919
s.SdkWorker().RegisterWorkflowWithOptions(workflow2Fn, workflow.RegisterOptions{Name: wt2})
920
921
// create
922
923
ctx := newContext(s.Context())
924
createTime := time.Now()
925
_, err := s.FrontendClient().CreateSchedule(ctx, req)
926
s.NoError(err)
927
928
// describe immediately after create and verify FutureActionTimes
929
describeRespAfterCreate, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
930
Namespace: s.Namespace().String(),
931
ScheduleId: sid,
932
})
933
s.NoError(err)
934
s.NotEmpty(describeRespAfterCreate.Info.FutureActionTimes, "FutureActionTimes should be set immediately after create")
935
// FutureActionTimes should be in the future (after createTime) and aligned to 5-second intervals
936
for i, fat := range describeRespAfterCreate.Info.FutureActionTimes {
937
s.True(fat.AsTime().After(createTime) || fat.AsTime().Equal(createTime),
938
"FutureActionTimes[%d] should be >= createTime", i)
939
s.Equal(int64(0), fat.AsTime().UnixNano()%int64(5*time.Second),
940
"FutureActionTimes[%d] should be aligned to 5-second intervals", i)
941
}
942
943
// sleep until we see two runs, plus a bit more to ensure that the second run has completed
944
s.Eventually(func() bool { return atomic.LoadInt32(&runs) == 2 }, 15*time.Second, 500*time.Millisecond)
945
time.Sleep(2 * time.Second) //nolint:forbidigo
946
947
// wait for visibility to stabilize on completed before calling describe,
948
// otherwise their recent actions may flake and differ
949
950
visibilityResponse := getScheduleEntryFromVisibility(s, sid, newContext, func(ent *schedulepb.ScheduleListEntry) bool {
951
recentActions := ent.GetInfo().GetRecentActions()
952
return len(recentActions) >= 2 && recentActions[1].GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
953
})
954
955
describeResp, err := s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
956
Namespace: s.Namespace().String(),
957
ScheduleId: sid,
958
})
959
s.NoError(err)
960
961
// validate describe response
962
963
checkSpec := func(spec *schedulepb.ScheduleSpec) {
964
protorequire.ProtoSliceEqual(s.T(), schedule.Spec.Interval, spec.Interval)
965
s.Nil(spec.Calendar)
966
s.Nil(spec.CronString)
967
s.ProtoElementsMatch([]*schedulepb.StructuredCalendarSpec{
968
{
969
Second: []*schedulepb.Range{{Start: 0, End: 0, Step: 1}},
970
Minute: []*schedulepb.Range{{Start: 11, End: 11, Step: 1}},
971
Hour: []*schedulepb.Range{{Start: 11, End: 23, Step: 11}},
972
DayOfMonth: []*schedulepb.Range{{Start: 11, End: 11, Step: 1}},
973
Month: []*schedulepb.Range{{Start: 11, End: 11, Step: 1}},
974
DayOfWeek: []*schedulepb.Range{{Start: 1, End: 1, Step: 1}},
975
Year: []*schedulepb.Range{{Start: 2011, End: 2011, Step: 1}},
976
},
977
{
978
Second: []*schedulepb.Range{{Start: 0, End: 0, Step: 1}},
979
Minute: []*schedulepb.Range{{Start: 0, End: 0, Step: 1}},
980
Hour: []*schedulepb.Range{{Start: 0, End: 0, Step: 1}},
981
DayOfMonth: []*schedulepb.Range{{Start: 10, End: 10, Step: 1}},
982
Month: []*schedulepb.Range{{Start: 1, End: 12, Step: 1}},
983
DayOfWeek: []*schedulepb.Range{{Start: 0, End: 6, Step: 1}},
984
Year: []*schedulepb.Range{{Start: 2010, End: 2010, Step: 1}},
985
},
986
}, spec.StructuredCalendar)
987
}
988
checkSpec(describeResp.Schedule.Spec)
989
990
s.Equal(enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, describeResp.Schedule.Policies.OverlapPolicy) // set to default value
991
s.EqualValues(365*24*3600, describeResp.Schedule.Policies.CatchupWindow.AsDuration().Seconds()) // set to default value
992
993
s.Equal(schSAValue.Data, describeResp.SearchAttributes.IndexedFields[csaKeyword].Data)
994
s.Equal(schSAIntValue.Data, describeResp.SearchAttributes.IndexedFields[csaInt].Data)
995
s.Equal(schSABoolValue.Data, describeResp.SearchAttributes.IndexedFields[csaBool].Data)
996
s.Nil(describeResp.SearchAttributes.IndexedFields[sadefs.BinaryChecksums])
997
s.Nil(describeResp.SearchAttributes.IndexedFields[sadefs.BuildIds])
998
s.Nil(describeResp.SearchAttributes.IndexedFields[sadefs.TemporalNamespaceDivision])
999
s.Equal(schMemo.Data, describeResp.Memo.Fields["schedmemo1"].Data)
1000
s.Equal(wfSAValue.Data, describeResp.Schedule.Action.GetStartWorkflow().SearchAttributes.IndexedFields[csaKeyword].Data)
1001
s.Equal(wfMemo.Data, describeResp.Schedule.Action.GetStartWorkflow().Memo.Fields["wfmemo1"].Data)
1002
1003
// GreaterOrEqual is used as we may have had other runs start while waiting for visibility
1004
durationNear(t, describeResp.Info.CreateTime.AsTime().Sub(createTime), 0)
1005
s.GreaterOrEqual(describeResp.Info.ActionCount, int64(2))
1006
s.EqualValues(0, describeResp.Info.MissedCatchupWindow)
1007
s.EqualValues(0, describeResp.Info.OverlapSkipped)
1008
s.GreaterOrEqual(len(describeResp.Info.RunningWorkflows), 0)
1009
s.GreaterOrEqual(len(describeResp.Info.RecentActions), 2)
1010
action0 := describeResp.Info.RecentActions[0]
1011
s.WithinRange(action0.ScheduleTime.AsTime(), createTime, time.Now())
1012
s.Equal(int64(0), action0.ScheduleTime.AsTime().UnixNano()%int64(5*time.Second))
1013
durationNear(t, action0.ActualTime.AsTime().Sub(action0.ScheduleTime.AsTime()), 0)
1014
1015
// validate list response
1016
1017
s.Equal(sid, visibilityResponse.ScheduleId)
1018
s.Equal(schSAValue.Data, visibilityResponse.SearchAttributes.IndexedFields[csaKeyword].Data)
1019
s.Equal(schSAIntValue.Data, describeResp.SearchAttributes.IndexedFields[csaInt].Data)
1020
s.Equal(schSABoolValue.Data, describeResp.SearchAttributes.IndexedFields[csaBool].Data)
1021
s.Nil(visibilityResponse.SearchAttributes.IndexedFields[sadefs.BinaryChecksums])
1022
s.Nil(visibilityResponse.SearchAttributes.IndexedFields[sadefs.BuildIds])
1023
s.Nil(visibilityResponse.SearchAttributes.IndexedFields[sadefs.TemporalNamespaceDivision])
1024
s.Equal(schMemo.Data, visibilityResponse.Memo.Fields["schedmemo1"].Data)
1025
checkSpec(visibilityResponse.Info.Spec)
1026
s.Equal(wt, visibilityResponse.Info.WorkflowType.Name)
1027
s.False(visibilityResponse.Info.Paused)
1028
assertSameRecentActions(s.T(), describeResp, visibilityResponse)
1029
assertRecentActionsNoDuplicateRunIDs(s.T(), describeResp.Info.RecentActions)
1030
1031
// list workflows
1032
1033
wfResp, err := s.FrontendClient().ListWorkflowExecutions(newContext(s.Context()), &workflowservice.ListWorkflowExecutionsRequest{
1034
Namespace: s.Namespace().String(),
1035
PageSize: 5,
1036
Query: "",
1037
})
1038
s.NoError(err)
1039
s.GreaterOrEqual(len(wfResp.Executions), 2) // could have had a 3rd run while waiting for visibility
1040
for _, ex := range wfResp.Executions {
1041
s.Equal(wt, ex.Type.Name, "should only see started workflows")
1042
}
1043
ex0 := wfResp.Executions[0]
1044
s.True(strings.HasPrefix(ex0.Execution.WorkflowId, wid))
1045
matchingRunId := false
1046
for _, recentAction := range describeResp.GetInfo().GetRecentActions() {
1047
if ex0.GetExecution().GetRunId() == recentAction.GetStartWorkflowResult().GetRunId() {
1048
matchingRunId = true
1049
break
1050
}
1051
}
1052
s.True(matchingRunId, "ListWorkflowExecutions returned a run ID wasn't in the describe response")
1053
s.Equal(wt, ex0.Type.Name)
1054
s.Nil(ex0.ParentExecution) // not a child workflow
1055
s.Equal(wfMemo.Data, ex0.Memo.Fields["wfmemo1"].Data)
1056
s.Equal(wfSAValue.Data, ex0.SearchAttributes.IndexedFields[csaKeyword].Data)
1057
s.Equal(payload.EncodeString(sid).Data, ex0.SearchAttributes.IndexedFields[sadefs.TemporalScheduledById].Data)
1058
var ex0StartTime time.Time
1059
s.NoError(payload.Decode(ex0.SearchAttributes.IndexedFields[sadefs.TemporalScheduledStartTime], &ex0StartTime))
1060
s.WithinRange(ex0StartTime, createTime, time.Now())
1061
s.Equal(int64(0), ex0StartTime.UnixNano()%int64(5*time.Second))
1062
1063
// list schedules with search attribute filter
1064
1065
listResp, err := s.FrontendClient().ListSchedules(newContext(s.Context()), &workflowservice.ListSchedulesRequest{
1066
Namespace: s.Namespace().String(),
1067
MaximumPageSize: 5,
1068
Query: "CustomKeywordField = 'schedule sa value' AND TemporalSchedulePaused = false",
1069
})
1070
s.NoError(err)
1071
s.Len(listResp.Schedules, 1)
1072
entry := listResp.Schedules[0]
1073
s.Equal(sid, entry.ScheduleId)
1074
1075
// list schedules with invalid search attribute filter
1076
1077
_, err = s.FrontendClient().ListSchedules(newContext(s.Context()), &workflowservice.ListSchedulesRequest{
1078
Namespace: s.Namespace().String(),
1079
MaximumPageSize: 5,
1080
Query: "ExecutionDuration > '1s'",
1081
})
1082
s.Error(err)
1083
1084
// update schedule, no updates to search attributes
1085
1086
schedule.Spec.Interval[0].Phase = durationpb.New(1 * time.Second)
1087
schedule.Action.GetStartWorkflow().WorkflowType.Name = wt2
1088
1089
updateTime := time.Now()
1090
_, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{
1091
Namespace: s.Namespace().String(),
1092
ScheduleId: sid,
1093
Schedule: schedule,
1094
Identity: "test",
1095
RequestId: uuid.NewString(),
1096
})
1097
s.NoError(err)
1098
1099
// wait for one new run
1100
s.Eventually(
1101
func() bool { return atomic.LoadInt32(&runs2) == 1 },
1102
7*time.Second,
1103
500*time.Millisecond,
1104
)
1105
1106
// describe again
1107
describeResp, err = s.FrontendClient().DescribeSchedule(
1108
newContext(s.Context()),
1109
&workflowservice.DescribeScheduleRequest{
1110
Namespace: s.Namespace().String(),
1111
ScheduleId: sid,
1112
},
1113
)
1114
s.NoError(err)
1115
1116
s.Len(describeResp.SearchAttributes.GetIndexedFields(), 3)
1117
s.Equal(schSAValue.Data, describeResp.SearchAttributes.IndexedFields[csaKeyword].Data)
1118
s.Equal(schSAIntValue.Data, describeResp.SearchAttributes.IndexedFields[csaInt].Data)
1119
s.Equal(schSABoolValue.Data, describeResp.SearchAttributes.IndexedFields[csaBool].Data)
1120
s.Equal(schMemo.Data, describeResp.Memo.Fields["schedmemo1"].Data)
1121
s.Equal(wfSAValue.Data, describeResp.Schedule.Action.GetStartWorkflow().SearchAttributes.IndexedFields[csaKeyword].Data)
1122
s.Equal(wfMemo.Data, describeResp.Schedule.Action.GetStartWorkflow().Memo.Fields["wfmemo1"].Data)
1123
1124
durationNear(t, describeResp.Info.UpdateTime.AsTime().Sub(updateTime), 0)
1125
lastAction := describeResp.Info.RecentActions[len(describeResp.Info.RecentActions)-1]
1126
s.Equal(int64(1000000000), lastAction.ScheduleTime.AsTime().UnixNano()%int64(5*time.Second), lastAction.ScheduleTime.AsTime().UnixNano())
1127
1128
// update schedule and search attributes
1129
1130
schedule.Spec.Interval[0].Phase = durationpb.New(1 * time.Second)
1131
schedule.Action.GetStartWorkflow().WorkflowType.Name = wt2
1132
1133
csaDouble := "CustomDoubleField"
1134
schSADoubleValue, _ := payload.Encode(3.14)
1135
schSAIntValue, _ = payload.Encode(321)
1136
_, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{
1137
Namespace: s.Namespace().String(),
1138
ScheduleId: sid,
1139
Schedule: schedule,
1140
Identity: "test",
1141
RequestId: uuid.NewString(),
1142
SearchAttributes: &commonpb.SearchAttributes{
1143
IndexedFields: map[string]*commonpb.Payload{
1144
csaKeyword: schSAValue, // same key, same value
1145
csaInt: schSAIntValue, // same key, new value
1146
csaDouble: schSADoubleValue, // new key
1147
// csaBool is removed
1148
},
1149
},
1150
})
1151
s.NoError(err)
1152
1153
// wait until search attributes are updated
1154
s.EventuallyWithT(
1155
func(c *assert.CollectT) {
1156
describeResp, err = s.FrontendClient().DescribeSchedule(
1157
newContext(s.Context()),
1158
&workflowservice.DescribeScheduleRequest{
1159
Namespace: s.Namespace().String(),
1160
ScheduleId: sid,
1161
},
1162
)
1163
require.NoError(c, err)
1164
require.Len(c, describeResp.SearchAttributes.GetIndexedFields(), 3)
1165
require.Equal(c, schSAValue.Data, describeResp.SearchAttributes.IndexedFields[csaKeyword].Data)
1166
require.Equal(c, schSAIntValue.Data, describeResp.SearchAttributes.IndexedFields[csaInt].Data)
1167
require.Equal(c, schSADoubleValue.Data, describeResp.SearchAttributes.IndexedFields[csaDouble].Data)
1168
require.NotContains(c, describeResp.SearchAttributes.IndexedFields, csaBool)
1169
},
1170
2*time.Second,
1171
500*time.Millisecond,
1172
)
1173
1174
// update schedule and unset search attributes
1175
1176
schedule.Spec.Interval[0].Phase = durationpb.New(1 * time.Second)
1177
schedule.Action.GetStartWorkflow().WorkflowType.Name = wt2
1178
1179
_, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{
1180
Namespace: s.Namespace().String(),
1181
ScheduleId: sid,
1182
Schedule: schedule,
1183
Identity: "test",
1184
RequestId: uuid.NewString(),
1185
SearchAttributes: &commonpb.SearchAttributes{},
1186
})
1187
s.NoError(err)
1188
1189
// wait until search attributes are updated
1190
s.EventuallyWithT(
1191
func(c *assert.CollectT) {
1192
describeResp, err = s.FrontendClient().DescribeSchedule(
1193
newContext(s.Context()),
1194
&workflowservice.DescribeScheduleRequest{
1195
Namespace: s.Namespace().String(),
1196
ScheduleId: sid,
1197
},
1198
)
1199
require.NoError(c, err)
1200
require.Empty(c, describeResp.SearchAttributes.GetIndexedFields())
1201
},
1202
5*time.Second,
1203
500*time.Millisecond,
1204
)
1205
1206
// pause
1207
1208
_, err = s.FrontendClient().PatchSchedule(newContext(s.Context()), &workflowservice.PatchScheduleRequest{
1209
Namespace: s.Namespace().String(),
1210
ScheduleId: sid,
1211
Patch: &schedulepb.SchedulePatch{
1212
Pause: "because I said so",
1213
},
1214
Identity: "test",
1215
RequestId: uuid.NewString(),
1216
})
1217
s.NoError(err)
1218
1219
time.Sleep(7 * time.Second) //nolint:forbidigo
1220
s.EqualValues(1, atomic.LoadInt32(&runs2), "has not run again")
1221
1222
describeResp, err = s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
1223
Namespace: s.Namespace().String(),
1224
ScheduleId: sid,
1225
})
1226
s.NoError(err)
1227
1228
s.True(describeResp.Schedule.State.Paused)
1229
s.Equal("because I said so", describeResp.Schedule.State.Notes)
1230
1231
// don't loop to wait for visibility, we already waited 7s from the patch
1232
listResp, err = s.FrontendClient().ListSchedules(newContext(s.Context()), &workflowservice.ListSchedulesRequest{
1233
Namespace: s.Namespace().String(),
1234
MaximumPageSize: 5,
1235
})
1236
s.NoError(err)
1237
s.Len(listResp.Schedules, 1)
1238
entry = listResp.Schedules[0]
1239
s.Equal(sid, entry.ScheduleId)
1240
s.True(entry.Info.Paused)
1241
s.Equal("because I said so", entry.Info.Notes)
1242
1243
// finally delete
1244
1245
_, err = s.FrontendClient().DeleteSchedule(newContext(s.Context()), &workflowservice.DeleteScheduleRequest{
1246
Namespace: s.Namespace().String(),
1247
ScheduleId: sid,
1248
Identity: "test",
1249
})
1250
s.NoError(err)
1251
1252
describeResp, err = s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
1253
Namespace: s.Namespace().String(),
1254
ScheduleId: sid,
1255
})
1256
var notFoundErr *serviceerror.NotFound
1257
s.ErrorAs(err, ¬FoundErr)
1258
1259
s.Eventually(func() bool { // wait for visibility
1260
listResp, err := s.FrontendClient().ListSchedules(newContext(s.Context()), &workflowservice.ListSchedulesRequest{
1261
Namespace: s.Namespace().String(),
1262
MaximumPageSize: 5,
1263
})
1264
s.NoError(err)
1265
return len(listResp.Schedules) == 0
1266
}, 10*time.Second, 1*time.Second)
1267
}
1268
1269
func testInput(t *testing.T, newContext contextFactory) {
1270
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
1271
1272
sid := "sched-test-input"
1273
wid := "sched-test-input-wf"
1274
wt := "sched-test-input-wt"
1275
1276
type myData struct {
1277
Stuff string
1278
Things []int
1279
}
1280
1281
input1 := &myData{
1282
Stuff: "here's some data",
1283
Things: []int{7, 8, 9},
1284
}
1285
input2 := map[int]float64{11: 1.4375}
1286
inputPayloads, err := payloads.Encode(input1, input2)
1287
s.NoError(err)
1288
1289
schedule := &schedulepb.Schedule{
1290
Spec: &schedulepb.ScheduleSpec{
1291
Interval: []*schedulepb.IntervalSpec{
1292
{Interval: durationpb.New(3 * time.Second)},
1293
},
1294
},
1295
Action: &schedulepb.ScheduleAction{
1296
Action: &schedulepb.ScheduleAction_StartWorkflow{
1297
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1298
WorkflowId: wid,
1299
WorkflowType: &commonpb.WorkflowType{Name: wt},
1300
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1301
Input: inputPayloads,
1302
},
1303
},
1304
},
1305
}
1306
req := &workflowservice.CreateScheduleRequest{
1307
Namespace: s.Namespace().String(),
1308
ScheduleId: sid,
1309
Schedule: schedule,
1310
Identity: "test",
1311
RequestId: uuid.NewString(),
1312
}
1313
1314
var runs int32
1315
workflowFn := func(ctx workflow.Context, arg1 *myData, arg2 map[int]float64) error {
1316
workflow.SideEffect(ctx, func(ctx workflow.Context) any {
1317
s.Equal(*input1, *arg1)
1318
s.Equal(input2, arg2)
1319
atomic.AddInt32(&runs, 1)
1320
return 0
1321
})
1322
return nil
1323
}
1324
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
1325
1326
ctx := newContext(s.Context())
1327
_, err = s.FrontendClient().CreateSchedule(ctx, req)
1328
s.NoError(err)
1329
1330
s.Eventually(func() bool { return atomic.LoadInt32(&runs) == 1 }, 8*time.Second, 200*time.Millisecond)
1331
}
1332
1333
func testLastCompletionAndError(t *testing.T, newContext contextFactory) {
1334
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
1335
1336
sid := "sched-test-last"
1337
wid := "sched-test-last-wf"
1338
wt := "sched-test-last-wt"
1339
1340
schedule := &schedulepb.Schedule{
1341
Spec: &schedulepb.ScheduleSpec{
1342
Interval: []*schedulepb.IntervalSpec{
1343
{Interval: durationpb.New(3 * time.Second)},
1344
},
1345
},
1346
Action: &schedulepb.ScheduleAction{
1347
Action: &schedulepb.ScheduleAction_StartWorkflow{
1348
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1349
WorkflowId: wid,
1350
WorkflowType: &commonpb.WorkflowType{Name: wt},
1351
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1352
},
1353
},
1354
},
1355
}
1356
req := &workflowservice.CreateScheduleRequest{
1357
Namespace: s.Namespace().String(),
1358
ScheduleId: sid,
1359
Schedule: schedule,
1360
Identity: "test",
1361
RequestId: uuid.NewString(),
1362
}
1363
1364
runs := make(map[string]struct{})
1365
var testComplete int32
1366
1367
workflowFn := func(ctx workflow.Context) (string, error) {
1368
var num int
1369
_ = workflow.SideEffect(ctx, func(ctx workflow.Context) any {
1370
runs[workflow.GetInfo(ctx).WorkflowExecution.ID] = struct{}{}
1371
return len(runs)
1372
}).Get(&num)
1373
1374
var lcr string
1375
if workflow.HasLastCompletionResult(ctx) {
1376
s.NoError(workflow.GetLastCompletionResult(ctx, &lcr))
1377
}
1378
1379
lastErr := workflow.GetLastError(ctx)
1380
1381
switch num {
1382
case 1:
1383
s.Empty(lcr)
1384
s.NoError(lastErr)
1385
return "this one succeeds", nil
1386
case 2:
1387
s.NoError(lastErr)
1388
s.Equal("this one succeeds", lcr)
1389
return "", errors.New("this one fails")
1390
case 3:
1391
s.Equal("this one succeeds", lcr)
1392
s.ErrorContains(lastErr, "this one fails")
1393
atomic.StoreInt32(&testComplete, 1)
1394
return "done", nil
1395
default:
1396
panic("shouldn't be running anymore")
1397
}
1398
}
1399
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
1400
1401
ctx := newContext(s.Context())
1402
_, err := s.FrontendClient().CreateSchedule(ctx, req)
1403
s.NoError(err)
1404
1405
s.Eventually(func() bool { return atomic.LoadInt32(&testComplete) == 1 }, 20*time.Second, 200*time.Millisecond)
1406
}
1407
1408
// testScheduleContinuesAfterWorkflowRetryFailure verifies a schedule keeps firing actions
1409
// after a scheduled workflow exhausts its retry policy and fails.
1410
func testScheduleContinuesAfterWorkflowRetryFailure(t *testing.T, newContext contextFactory) {
1411
// Recording FAILED actions across the workflow's retry chain relies on the scheduler matching
1412
// completions by the request ID carried in the completion callback token (which survives the new
1413
// runs created by retries). That requires the envelope token format, which is gated off by default
1414
// for safe rollout, so enable it explicitly here.
1415
opts := append(scheduleCommonOpts(t), testcore.WithDynamicConfig(callback.EncodeInternalTokenWithEnvelope, true))
1416
s := newScheduleEnv(t, opts...)
1417
1418
sid := testcore.RandomizeStr("sched-retry-fail")
1419
wid := testcore.RandomizeStr("sched-retry-fail-wf")
1420
wt := testcore.RandomizeStr("sched-retry-fail-wt")
1421
1422
var sawRetry int32
1423
workflowFn := func(ctx workflow.Context) error {
1424
if workflow.GetInfo(ctx).Attempt > 1 {
1425
atomic.StoreInt32(&sawRetry, 1)
1426
}
1427
return errors.New("intentional failure to force a retry")
1428
}
1429
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
1430
1431
schedule := &schedulepb.Schedule{
1432
Spec: &schedulepb.ScheduleSpec{
1433
Interval: []*schedulepb.IntervalSpec{
1434
{Interval: durationpb.New(3 * time.Second)},
1435
},
1436
},
1437
Action: &schedulepb.ScheduleAction{
1438
Action: &schedulepb.ScheduleAction_StartWorkflow{
1439
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1440
WorkflowId: wid,
1441
WorkflowType: &commonpb.WorkflowType{Name: wt},
1442
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1443
RetryPolicy: &commonpb.RetryPolicy{
1444
InitialInterval: durationpb.New(1 * time.Second),
1445
BackoffCoefficient: 1.0,
1446
MaximumAttempts: 2,
1447
},
1448
},
1449
},
1450
},
1451
}
1452
1453
ctx := newContext(s.Context())
1454
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
1455
Namespace: s.Namespace().String(),
1456
ScheduleId: sid,
1457
Schedule: schedule,
1458
Identity: "test",
1459
RequestId: uuid.NewString(),
1460
})
1461
require.NoError(t, err)
1462
1463
// Two FAILED actions proves the schedule kept firing past the first retry-failure.
1464
var failedActions int
1465
var lastDescribe *workflowservice.DescribeScheduleResponse
1466
s.Eventually(func() bool {
1467
desc, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
1468
Namespace: s.Namespace().String(),
1469
ScheduleId: sid,
1470
})
1471
if descErr != nil {
1472
return false
1473
}
1474
lastDescribe = desc
1475
failedActions = 0
1476
for _, a := range desc.GetInfo().GetRecentActions() {
1477
if a.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_FAILED {
1478
failedActions++
1479
}
1480
}
1481
return atomic.LoadInt32(&sawRetry) == 1 && failedActions >= 2
1482
}, 30*time.Second, 500*time.Millisecond,
1483
"schedule should keep recording FAILED actions after the workflow retry-fails")
1484
1485
s.Equal(int32(1), atomic.LoadInt32(&sawRetry), "scheduled workflow should have retried (attempt > 1)")
1486
s.GreaterOrEqual(failedActions, 2, "schedule should record multiple retry-failed actions")
1487
s.GreaterOrEqual(lastDescribe.GetInfo().GetActionCount(), int64(2))
1488
s.False(lastDescribe.GetSchedule().GetState().GetPaused(), "a retry-failed workflow must not pause the schedule")
1489
}
1490
1491
// testScheduledWorkflowContinueAsNewCompletion validates that the CHASM scheduler observes the
1492
// completion of a scheduled workflow that continues-as-new before completing. The scheduler matches
1493
// completions by the request ID on the Nexus completion callback it attaches at start; if that ID is
1494
// lost across continue-as-new the completion is silently dropped, so with a buffering overlap policy
1495
// the scheduler believes the action is still running and never starts the buffered actions.
1496
//
1497
// It asserts the schedule records several COMPLETED actions (the buffer only drains as completions
1498
// are observed) and that the completion callback is written intact into both the original and the
1499
// continued-as-new run. CHASM-only: V1 uses no Nexus completion callbacks.
1500
func testScheduledWorkflowContinueAsNewCompletion(t *testing.T, newContext contextFactory) {
1501
// The scheduler matches the continued-as-new run's completion by the request ID carried in the
1502
// completion callback token, which only survives continue-as-new in the envelope token format.
1503
// That format is gated off by default for safe rollout, so enable it explicitly here.
1504
opts := append(scheduleCommonOpts(t), testcore.WithDynamicConfig(callback.EncodeInternalTokenWithEnvelope, true))
1505
s := newScheduleEnv(t, opts...)
1506
1507
sid := testcore.RandomizeStr("sched-can-completion")
1508
wid := testcore.RandomizeStr("sched-can-completion-wf")
1509
wt := testcore.RandomizeStr("sched-can-completion-wt")
1510
1511
// Continue-as-new once, then complete: the completion is delivered from the continued run,
1512
// exercising completion-callback propagation across continue-as-new.
1513
workflowFn := func(ctx workflow.Context) error {
1514
if workflow.GetInfo(ctx).ContinuedExecutionRunID == "" {
1515
return workflow.NewContinueAsNewError(ctx, wt)
1516
}
1517
return nil
1518
}
1519
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
1520
1521
schedule := &schedulepb.Schedule{
1522
Spec: &schedulepb.ScheduleSpec{
1523
Interval: []*schedulepb.IntervalSpec{
1524
{Interval: durationpb.New(1 * time.Second)},
1525
},
1526
},
1527
// BUFFER_ALL gates each start on the previous action completing. If a completion is dropped
1528
// because request ID is not carried over on continue-as-newthe scheduler believes the action
1529
// is still running and the buffered actions never start, so only the first action ever completes.
1530
Policies: &schedulepb.SchedulePolicies{
1531
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
1532
},
1533
Action: &schedulepb.ScheduleAction{
1534
Action: &schedulepb.ScheduleAction_StartWorkflow{
1535
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1536
WorkflowId: wid,
1537
WorkflowType: &commonpb.WorkflowType{Name: wt},
1538
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1539
},
1540
},
1541
},
1542
}
1543
req := &workflowservice.CreateScheduleRequest{
1544
Namespace: s.Namespace().String(),
1545
ScheduleId: sid,
1546
Schedule: schedule,
1547
Identity: "test",
1548
RequestId: uuid.NewString(),
1549
}
1550
1551
ctx := newContext(s.Context())
1552
_, err := s.FrontendClient().CreateSchedule(ctx, req)
1553
require.NoError(t, err)
1554
1555
// getCompletionCallback returns the Nexus completion callback the scheduler attached, found in the
1556
// run's WorkflowExecutionStarted event.
1557
getCompletionCallback := func(events []*historypb.HistoryEvent) *commonpb.Callback {
1558
for _, e := range events {
1559
for _, cb := range e.GetWorkflowExecutionStartedEventAttributes().GetCompletionCallbacks() {
1560
if cb.GetNexus().GetUrl() == chasm.NexusCompletionHandlerURL {
1561
return cb
1562
}
1563
}
1564
}
1565
return nil
1566
}
1567
1568
// The scheduler only starts the next buffered action after observing the previous one complete,
1569
// so it records multiple COMPLETED actions only if continue-as-new completions are delivered. With
1570
// the bug it stalls after the first. (StartWorkflowStatus is set solely from the completion
1571
// callback the scheduler records, not from visibility.)
1572
const wantCompleted = 3
1573
var completedWFID string
1574
s.Eventually(func() bool {
1575
desc, descErr := s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
1576
Namespace: s.Namespace().String(),
1577
ScheduleId: sid,
1578
})
1579
if descErr != nil {
1580
return false
1581
}
1582
completed := 0
1583
for _, a := range desc.GetInfo().GetRecentActions() {
1584
if a.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED {
1585
completed++
1586
completedWFID = a.GetStartWorkflowResult().GetWorkflowId()
1587
}
1588
}
1589
return completed >= wantCompleted
1590
}, 15*time.Second, 200*time.Millisecond,
1591
"scheduler should record %d completed actions", wantCompleted)
1592
1593
// Verify the completion callback was written into both runs of a completed action: the
1594
// continued-as-new run (latest) and the original run it continued from. The header (callback
1595
// token) must be identical, confirming the callback was propagated intact across continue-as-new.
1596
s.NotEmpty(completedWFID)
1597
canHist := s.GetHistory(s.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: completedWFID})
1598
canCB := getCompletionCallback(canHist)
1599
s.NotNil(canCB, "continued-as-new run must carry the completion callback")
1600
1601
var continuedFromRunID string
1602
for _, e := range canHist {
1603
if a := e.GetWorkflowExecutionStartedEventAttributes(); a != nil {
1604
continuedFromRunID = a.GetContinuedExecutionRunId()
1605
break
1606
}
1607
}
1608
s.NotEmpty(continuedFromRunID, "completed action should have continued-as-new")
1609
1610
origHist := s.GetHistory(s.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: completedWFID, RunId: continuedFromRunID})
1611
origCB := getCompletionCallback(origHist)
1612
s.NotNil(origCB, "original run must carry the completion callback")
1613
s.Equal(origCB.GetNexus().GetHeader(), canCB.GetNexus().GetHeader(), "completion callback must be propagated intact across continue-as-new")
1614
}
1615
1616
func testListSchedulesReturnsWorkflowStatus(t *testing.T, newContext contextFactory) {
1617
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
1618
1619
sid := "sched-test-list-running"
1620
wid := "sched-test-list-running-wf"
1621
wt := "sched-test-list-running-wt"
1622
1623
// Set up a schedule that immediately starts a single running workflow
1624
schedule := &schedulepb.Schedule{
1625
Spec: &schedulepb.ScheduleSpec{
1626
Interval: []*schedulepb.IntervalSpec{
1627
{Interval: durationpb.New(3 * time.Second)},
1628
},
1629
},
1630
Action: &schedulepb.ScheduleAction{
1631
Action: &schedulepb.ScheduleAction_StartWorkflow{
1632
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1633
WorkflowId: wid,
1634
WorkflowType: &commonpb.WorkflowType{Name: wt},
1635
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1636
},
1637
},
1638
},
1639
}
1640
patch := &schedulepb.SchedulePatch{
1641
TriggerImmediately: &schedulepb.TriggerImmediatelyRequest{},
1642
}
1643
1644
// The workflow sits open until we've asserted it can be listed as running
1645
resumeSignal := "resume"
1646
workflowFn := func(ctx workflow.Context) error {
1647
selector := workflow.NewSelector(ctx)
1648
selector.AddReceive(workflow.GetSignalChannel(ctx, resumeSignal), func(c workflow.ReceiveChannel, more bool) {
1649
// nothing to do
1650
})
1651
selector.Select(ctx)
1652
return nil
1653
}
1654
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
1655
1656
req := &workflowservice.CreateScheduleRequest{
1657
Namespace: s.Namespace().String(),
1658
ScheduleId: sid,
1659
Schedule: schedule,
1660
InitialPatch: patch,
1661
RequestId: uuid.NewString(),
1662
}
1663
ctx := newContext(s.Context())
1664
_, err := s.FrontendClient().CreateSchedule(ctx, req)
1665
s.NoError(err)
1666
1667
// validate RecentActions made it to visibility
1668
listResp := getScheduleEntryFromVisibility(s, sid, newContext, func(listResp *schedulepb.ScheduleListEntry) bool {
1669
return len(listResp.Info.RecentActions) >= 1
1670
})
1671
s.Len(listResp.Info.RecentActions, 1)
1672
1673
a1 := listResp.Info.RecentActions[0]
1674
s.True(strings.HasPrefix(a1.StartWorkflowResult.WorkflowId, wid))
1675
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, a1.StartWorkflowStatus)
1676
1677
// let the started workflow complete
1678
_, err = s.FrontendClient().SignalWorkflowExecution(newContext(s.Context()), &workflowservice.SignalWorkflowExecutionRequest{
1679
Namespace: s.Namespace().String(),
1680
WorkflowExecution: &commonpb.WorkflowExecution{
1681
WorkflowId: a1.StartWorkflowResult.WorkflowId,
1682
RunId: a1.StartWorkflowResult.RunId,
1683
},
1684
SignalName: resumeSignal,
1685
})
1686
s.NoError(err)
1687
1688
// now wait for second recent action to land in visbility
1689
listResp = getScheduleEntryFromVisibility(s, sid, newContext, func(listResp *schedulepb.ScheduleListEntry) bool {
1690
return len(listResp.Info.RecentActions) >= 2
1691
})
1692
1693
a1 = listResp.Info.RecentActions[0]
1694
a2 := listResp.Info.RecentActions[1]
1695
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, a1.StartWorkflowStatus)
1696
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, a2.StartWorkflowStatus)
1697
1698
// Also verify that DescribeSchedule's output matches
1699
descResp, err := s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
1700
Namespace: s.Namespace().String(),
1701
ScheduleId: sid,
1702
})
1703
s.NoError(err)
1704
assertSameRecentActions(s.T(), descResp, listResp)
1705
1706
// Verify no duplicate RunIds in recent actions (regression for migration dedup bug).
1707
assertRecentActionsNoDuplicateRunIDs(s.T(), descResp.Info.RecentActions)
1708
assertRecentActionsNoDuplicateRunIDs(s.T(), listResp.Info.RecentActions)
1709
}
1710
1711
func testUpdateIntervalTakesEffect(t *testing.T, newContext contextFactory) {
1712
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
1713
1714
sid := "sched-test-update-interval"
1715
wid := "sched-test-update-interval-wf"
1716
wt := "sched-test-update-interval-wt"
1717
1718
var runs int32
1719
workflowFn := func(ctx workflow.Context) error {
1720
workflow.SideEffect(ctx, func(ctx workflow.Context) any {
1721
atomic.AddInt32(&runs, 1)
1722
return 0
1723
})
1724
return nil
1725
}
1726
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
1727
1728
// Create schedule with a long interval (300s) - won't fire for 5 minutes.
1729
schedule := &schedulepb.Schedule{
1730
Spec: &schedulepb.ScheduleSpec{
1731
Interval: []*schedulepb.IntervalSpec{
1732
{Interval: durationpb.New(300 * time.Second)},
1733
},
1734
},
1735
Action: &schedulepb.ScheduleAction{
1736
Action: &schedulepb.ScheduleAction_StartWorkflow{
1737
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1738
WorkflowId: wid,
1739
WorkflowType: &commonpb.WorkflowType{Name: wt},
1740
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1741
},
1742
},
1743
},
1744
}
1745
1746
ctx := newContext(s.Context())
1747
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
1748
Namespace: s.Namespace().String(),
1749
ScheduleId: sid,
1750
Schedule: schedule,
1751
Identity: "test",
1752
RequestId: uuid.NewString(),
1753
})
1754
s.NoError(err)
1755
1756
// Update the interval to be very short (1s).
1757
schedule.Spec.Interval[0].Interval = durationpb.New(1 * time.Second)
1758
_, err = s.FrontendClient().UpdateSchedule(ctx, &workflowservice.UpdateScheduleRequest{
1759
Namespace: s.Namespace().String(),
1760
ScheduleId: sid,
1761
Schedule: schedule,
1762
Identity: "test",
1763
RequestId: uuid.NewString(),
1764
})
1765
s.NoError(err)
1766
1767
// After updating to 1s interval, we should see runs start within a few seconds.
1768
s.Eventually(
1769
func() bool { return atomic.LoadInt32(&runs) >= 2 },
1770
10*time.Second,
1771
500*time.Millisecond,
1772
"expected at least 2 runs within 10s after updating interval to 1s",
1773
)
1774
}
1775
1776
func testListScheduleMatchingTimes(t *testing.T, newContext contextFactory) {
1777
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
1778
1779
sid := "sched-test-list-matching-times"
1780
1781
schedule := &schedulepb.Schedule{
1782
Spec: &schedulepb.ScheduleSpec{
1783
Interval: []*schedulepb.IntervalSpec{
1784
{Interval: durationpb.New(1 * time.Hour)},
1785
},
1786
},
1787
Action: &schedulepb.ScheduleAction{
1788
Action: &schedulepb.ScheduleAction_StartWorkflow{
1789
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1790
WorkflowId: "wf-list-matching-times",
1791
WorkflowType: &commonpb.WorkflowType{Name: "action"},
1792
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1793
},
1794
},
1795
},
1796
}
1797
req := &workflowservice.CreateScheduleRequest{
1798
Namespace: s.Namespace().String(),
1799
ScheduleId: sid,
1800
Schedule: schedule,
1801
Identity: "test",
1802
RequestId: uuid.NewString(),
1803
}
1804
1805
ctx := newContext(s.Context())
1806
_, err := s.FrontendClient().CreateSchedule(ctx, req)
1807
s.NoError(err)
1808
1809
// Query for matching times over a 5-hour window.
1810
now := time.Now().UTC().Truncate(time.Hour).Add(time.Hour) // Start of next hour
1811
startTime := timestamppb.New(now)
1812
endTime := timestamppb.New(now.Add(5 * time.Hour))
1813
1814
resp, err := s.FrontendClient().ListScheduleMatchingTimes(ctx, &workflowservice.ListScheduleMatchingTimesRequest{
1815
Namespace: s.Namespace().String(),
1816
ScheduleId: sid,
1817
StartTime: startTime,
1818
EndTime: endTime,
1819
})
1820
s.NoError(err)
1821
// With 1-hour interval over 5 hours, we expect 5 matching times.
1822
s.Len(resp.GetStartTime(), 5)
1823
}
1824
1825
func testLimitMemoSpecSize(t *testing.T, newContext contextFactory) {
1826
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
1827
1828
expectedLimit := scheduler.CurrentTweakablePolicies.SpecFieldLengthLimit
1829
1830
sid := "sched-test-limit-memo-size"
1831
wid := "sched-test-limit-memo-size-wf"
1832
wt := "sched-test-limit-memo-size-wt"
1833
1834
schedule := &schedulepb.Schedule{
1835
Spec: &schedulepb.ScheduleSpec{},
1836
Action: &schedulepb.ScheduleAction{
1837
Action: &schedulepb.ScheduleAction_StartWorkflow{
1838
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1839
WorkflowId: wid,
1840
WorkflowType: &commonpb.WorkflowType{Name: wt},
1841
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1842
},
1843
},
1844
},
1845
}
1846
1847
// Set up a schedule with a large number of spec items that should be trimmed in
1848
// the memo block.
1849
for i := 0; i < expectedLimit*2; i++ {
1850
schedule.Spec.Interval = append(schedule.Spec.Interval, &schedulepb.IntervalSpec{
1851
Interval: durationpb.New(time.Duration(i+1) * time.Second),
1852
})
1853
schedule.Spec.StructuredCalendar = append(schedule.Spec.StructuredCalendar, &schedulepb.StructuredCalendarSpec{
1854
Minute: []*schedulepb.Range{
1855
{
1856
Start: int32(i + 1),
1857
End: int32(i + 1),
1858
},
1859
},
1860
})
1861
schedule.Spec.ExcludeStructuredCalendar = append(schedule.Spec.ExcludeStructuredCalendar, &schedulepb.StructuredCalendarSpec{
1862
Second: []*schedulepb.Range{
1863
{
1864
Start: int32(i + 1),
1865
End: int32(i + 1),
1866
},
1867
},
1868
})
1869
}
1870
1871
// Create the schedule.
1872
req := &workflowservice.CreateScheduleRequest{
1873
Namespace: s.Namespace().String(),
1874
ScheduleId: sid,
1875
Schedule: schedule,
1876
Identity: "test",
1877
RequestId: uuid.NewString(),
1878
}
1879
s.SdkWorker().RegisterWorkflowWithOptions(
1880
func(ctx workflow.Context) error { return nil },
1881
workflow.RegisterOptions{Name: wt},
1882
)
1883
ctx := newContext(s.Context())
1884
_, err := s.FrontendClient().CreateSchedule(ctx, req)
1885
s.NoError(err)
1886
1887
// Verify the memo field length limit was enforced.
1888
entry := getScheduleEntryFromVisibility(s, sid, newContext, nil)
1889
require.NotNil(t, entry)
1890
spec := entry.GetInfo().GetSpec()
1891
require.Len(t, spec.GetInterval(), expectedLimit)
1892
require.Len(t, spec.GetStructuredCalendar(), expectedLimit)
1893
require.Len(t, spec.GetExcludeStructuredCalendar(), expectedLimit)
1894
}
1895
1896
func testCountSchedules(t *testing.T, newContext contextFactory) {
1897
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
1898
1899
// Create multiple schedules with different paused states
1900
sidPrefix := "sched-test-count-"
1901
wid := "sched-test-count-wf"
1902
wt := "sched-test-count-wt"
1903
1904
// Create 3 schedules: 2 active, 1 paused
1905
for i := range 3 {
1906
sid := fmt.Sprintf("%s%d", sidPrefix, i)
1907
paused := i == 2 // Third schedule is paused
1908
1909
schedule := &schedulepb.Schedule{
1910
Spec: &schedulepb.ScheduleSpec{
1911
Interval: []*schedulepb.IntervalSpec{
1912
{Interval: durationpb.New(1 * time.Hour)},
1913
},
1914
},
1915
Action: &schedulepb.ScheduleAction{
1916
Action: &schedulepb.ScheduleAction_StartWorkflow{
1917
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1918
WorkflowId: fmt.Sprintf("%s-%d", wid, i),
1919
WorkflowType: &commonpb.WorkflowType{Name: wt},
1920
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1921
},
1922
},
1923
},
1924
State: &schedulepb.ScheduleState{
1925
Paused: paused,
1926
},
1927
}
1928
1929
_, err := s.FrontendClient().CreateSchedule(newContext(s.Context()), &workflowservice.CreateScheduleRequest{
1930
Namespace: s.Namespace().String(),
1931
ScheduleId: sid,
1932
Schedule: schedule,
1933
Identity: "test",
1934
RequestId: uuid.NewString(),
1935
})
1936
s.NoError(err)
1937
}
1938
1939
// Test basic count (all schedules)
1940
s.Eventually(func() bool {
1941
countResp, err := s.FrontendClient().CountSchedules(newContext(s.Context()), &workflowservice.CountSchedulesRequest{
1942
Namespace: s.Namespace().String(),
1943
})
1944
return err == nil && countResp.Count >= 3
1945
}, 15*time.Second, 1*time.Second, "Expected at least 3 schedules")
1946
1947
// Test count with query filter for paused schedules
1948
s.Eventually(func() bool {
1949
countResp, err := s.FrontendClient().CountSchedules(newContext(s.Context()), &workflowservice.CountSchedulesRequest{
1950
Namespace: s.Namespace().String(),
1951
Query: fmt.Sprintf("%s = true", sadefs.TemporalSchedulePaused),
1952
})
1953
return err == nil && countResp.Count >= 1
1954
}, 15*time.Second, 1*time.Second, "Expected at least 1 paused schedule")
1955
1956
// Test count with query filter for non-paused schedules
1957
s.Eventually(func() bool {
1958
countResp, err := s.FrontendClient().CountSchedules(newContext(s.Context()), &workflowservice.CountSchedulesRequest{
1959
Namespace: s.Namespace().String(),
1960
Query: fmt.Sprintf("%s = false", sadefs.TemporalSchedulePaused),
1961
})
1962
return err == nil && countResp.Count >= 2
1963
}, 15*time.Second, 1*time.Second, "Expected at least 2 non-paused schedules")
1964
}
1965
1966
func testListSchedulesPagination(t *testing.T, newContext contextFactory) {
1967
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
1968
1969
const numSchedules = 4
1970
sidPrefix := "sched-test-pagination-"
1971
1972
for i := range numSchedules {
1973
sid := fmt.Sprintf("%s%d", sidPrefix, i)
1974
schedule := &schedulepb.Schedule{
1975
Spec: &schedulepb.ScheduleSpec{
1976
Interval: []*schedulepb.IntervalSpec{
1977
{Interval: durationpb.New(1 * time.Hour)},
1978
},
1979
},
1980
Action: &schedulepb.ScheduleAction{
1981
Action: &schedulepb.ScheduleAction_StartWorkflow{
1982
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
1983
WorkflowId: fmt.Sprintf("wf-pagination-%d", i),
1984
WorkflowType: &commonpb.WorkflowType{Name: "action"},
1985
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
1986
},
1987
},
1988
},
1989
}
1990
_, err := s.FrontendClient().CreateSchedule(newContext(s.Context()), &workflowservice.CreateScheduleRequest{
1991
Namespace: s.Namespace().String(),
1992
ScheduleId: sid,
1993
Schedule: schedule,
1994
Identity: "test",
1995
RequestId: uuid.NewString(),
1996
})
1997
s.NoError(err)
1998
}
1999
2000
// Wait for all schedules to be visible.
2001
s.Eventually(func() bool {
2002
countResp, err := s.FrontendClient().CountSchedules(newContext(s.Context()), &workflowservice.CountSchedulesRequest{
2003
Namespace: s.Namespace().String(),
2004
})
2005
return err == nil && countResp.Count >= numSchedules
2006
}, 15*time.Second, 1*time.Second, "Expected all schedules to be visible")
2007
2008
// Paginate with page size 2 and collect all schedule IDs.
2009
ctx := newContext(s.Context())
2010
var allIDs []string
2011
var nextPageToken []byte
2012
for {
2013
resp, err := s.FrontendClient().ListSchedules(ctx, &workflowservice.ListSchedulesRequest{
2014
Namespace: s.Namespace().String(),
2015
MaximumPageSize: 2,
2016
NextPageToken: nextPageToken,
2017
})
2018
s.NoError(err)
2019
for _, entry := range resp.Schedules {
2020
allIDs = append(allIDs, entry.ScheduleId)
2021
}
2022
nextPageToken = resp.NextPageToken
2023
if len(nextPageToken) == 0 {
2024
break
2025
}
2026
// Each page except possibly the last should have entries.
2027
s.NotEmpty(resp.Schedules)
2028
}
2029
2030
// Verify we found all created schedules.
2031
for i := range numSchedules {
2032
sid := fmt.Sprintf("%s%d", sidPrefix, i)
2033
s.Contains(allIDs, sid, "Expected schedule %s in paginated results", sid)
2034
}
2035
}
2036
2037
func testListSchedulesFilterAndEntryFields(t *testing.T, newContext contextFactory) {
2038
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
2039
2040
sid := "sched-test-list-fields"
2041
wt := "sched-test-list-fields-wt"
2042
2043
schMemo, _ := payload.Encode("memo value")
2044
csaKeyword := "CustomKeywordField"
2045
schSAValue, _ := payload.Encode("sa-val")
2046
2047
// Create a paused schedule with memo and custom search attributes.
2048
schedule := &schedulepb.Schedule{
2049
Spec: &schedulepb.ScheduleSpec{
2050
Interval: []*schedulepb.IntervalSpec{
2051
{Interval: durationpb.New(1 * time.Hour)},
2052
},
2053
},
2054
Action: &schedulepb.ScheduleAction{
2055
Action: &schedulepb.ScheduleAction_StartWorkflow{
2056
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2057
WorkflowId: "wf-" + sid,
2058
WorkflowType: &commonpb.WorkflowType{Name: wt},
2059
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2060
},
2061
},
2062
},
2063
State: &schedulepb.ScheduleState{
2064
Paused: true,
2065
Notes: "paused for test",
2066
},
2067
}
2068
2069
ctx := newContext(s.Context())
2070
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
2071
Namespace: s.Namespace().String(),
2072
ScheduleId: sid,
2073
Schedule: schedule,
2074
Identity: "test",
2075
RequestId: uuid.NewString(),
2076
Memo: &commonpb.Memo{
2077
Fields: map[string]*commonpb.Payload{
2078
"schedmemo1": schMemo,
2079
},
2080
},
2081
SearchAttributes: &commonpb.SearchAttributes{
2082
IndexedFields: map[string]*commonpb.Payload{
2083
csaKeyword: schSAValue,
2084
},
2085
},
2086
})
2087
s.NoError(err)
2088
2089
// Wait for the schedule to appear with correct paused state.
2090
entry := getScheduleEntryFromVisibility(s, sid, newContext, func(e *schedulepb.ScheduleListEntry) bool {
2091
return e.Info.Paused
2092
})
2093
2094
// Verify entry-level fields.
2095
s.Equal(schMemo.Data, entry.Memo.Fields["schedmemo1"].Data)
2096
s.Equal(schSAValue.Data, entry.SearchAttributes.IndexedFields[csaKeyword].Data)
2097
s.Equal(wt, entry.Info.WorkflowType.Name)
2098
s.True(entry.Info.Paused)
2099
s.Equal("paused for test", entry.Info.Notes)
2100
2101
// Filter by TemporalSchedulePaused should find this schedule.
2102
s.EventuallyWithT(func(c *assert.CollectT) {
2103
listResp, err := s.FrontendClient().ListSchedules(ctx, &workflowservice.ListSchedulesRequest{
2104
Namespace: s.Namespace().String(),
2105
MaximumPageSize: 10,
2106
Query: fmt.Sprintf("%s = true", sadefs.TemporalSchedulePaused),
2107
})
2108
require.NoError(c, err)
2109
var ids []string
2110
for _, e := range listResp.Schedules {
2111
ids = append(ids, e.ScheduleId)
2112
}
2113
require.Contains(c, ids, sid)
2114
}, 15*time.Second, 1*time.Second)
2115
2116
// Filter for paused=false should not include this schedule.
2117
s.EventuallyWithT(func(c *assert.CollectT) {
2118
listResp, err := s.FrontendClient().ListSchedules(ctx, &workflowservice.ListSchedulesRequest{
2119
Namespace: s.Namespace().String(),
2120
MaximumPageSize: 10,
2121
Query: fmt.Sprintf("%s = false", sadefs.TemporalSchedulePaused),
2122
})
2123
require.NoError(c, err)
2124
for _, e := range listResp.Schedules {
2125
require.NotEqual(c, sid, e.ScheduleId)
2126
}
2127
}, 15*time.Second, 1*time.Second)
2128
}
2129
2130
func testListSchedulesFilterByScheduleID(t *testing.T, newContext contextFactory) {
2131
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
2132
2133
sid1 := "sched-filter-by-id-alpha"
2134
sid2 := "sched-filter-by-id-beta"
2135
2136
schedule := func(sid string) *schedulepb.Schedule {
2137
return &schedulepb.Schedule{
2138
Spec: &schedulepb.ScheduleSpec{
2139
Interval: []*schedulepb.IntervalSpec{
2140
{Interval: durationpb.New(1 * time.Hour)},
2141
},
2142
},
2143
Action: &schedulepb.ScheduleAction{
2144
Action: &schedulepb.ScheduleAction_StartWorkflow{
2145
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2146
WorkflowId: "wf-" + sid,
2147
WorkflowType: &commonpb.WorkflowType{Name: "action"},
2148
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2149
},
2150
},
2151
},
2152
State: &schedulepb.ScheduleState{Paused: true},
2153
}
2154
}
2155
2156
ctx := newContext(s.Context())
2157
2158
// Create two schedules.
2159
for _, sid := range []string{sid1, sid2} {
2160
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
2161
Namespace: s.Namespace().String(),
2162
ScheduleId: sid,
2163
Schedule: schedule(sid),
2164
Identity: "test",
2165
RequestId: uuid.NewString(),
2166
})
2167
s.NoError(err)
2168
}
2169
2170
// Wait for both schedules to appear in visibility.
2171
getScheduleEntryFromVisibility(s, sid1, newContext, nil)
2172
getScheduleEntryFromVisibility(s, sid2, newContext, nil)
2173
2174
listScheduleIDs := func(query string) []string {
2175
t.Helper()
2176
listResp, err := s.FrontendClient().ListSchedules(ctx, &workflowservice.ListSchedulesRequest{
2177
Namespace: s.Namespace().String(),
2178
MaximumPageSize: 10,
2179
Query: query,
2180
})
2181
require.NoError(t, err)
2182
var ids []string
2183
for _, e := range listResp.Schedules {
2184
ids = append(ids, e.ScheduleId)
2185
}
2186
return ids
2187
}
2188
2189
// wantIDs is the exact set of schedule IDs expected in the result.
2190
// IsNegativeScheduleIDOperator drives whether an operator excludes or includes:
2191
// negative operators (!=, NOT IN, NOT STARTS_WITH) produce AND in the rewriter so both
2192
// V1 and V2 forms are excluded; positive operators produce OR so both forms are included.
2193
tests := []struct {
2194
name string
2195
query string
2196
wantIDs []string
2197
}{
2198
{
2199
name: "Equal",
2200
query: fmt.Sprintf("ScheduleId = '%s'", sid1),
2201
wantIDs: []string{sid1},
2202
},
2203
{
2204
// scheduler.IsNegativeScheduleIDOperator("!=") == true
2205
name: "NotEqual",
2206
query: fmt.Sprintf("ScheduleId != '%s'", sid1),
2207
wantIDs: []string{sid2},
2208
},
2209
{
2210
name: "StartsWith",
2211
query: "ScheduleId STARTS_WITH 'sched-filter-by-id-'",
2212
wantIDs: []string{sid1, sid2},
2213
},
2214
{
2215
name: "StartsWithSpecific",
2216
query: "ScheduleId STARTS_WITH 'sched-filter-by-id-a'",
2217
wantIDs: []string{sid1},
2218
},
2219
{
2220
// scheduler.IsNegativeScheduleIDOperator("not starts_with") == true
2221
name: "NotStartsWith",
2222
query: "ScheduleId NOT STARTS_WITH 'sched-filter-by-id-a'",
2223
wantIDs: []string{sid2},
2224
},
2225
{
2226
name: "In",
2227
query: fmt.Sprintf("ScheduleId IN ('%s', '%s')", sid1, sid2),
2228
wantIDs: []string{sid1, sid2},
2229
},
2230
{
2231
name: "InSingle",
2232
query: fmt.Sprintf("ScheduleId IN ('%s')", sid2),
2233
wantIDs: []string{sid2},
2234
},
2235
{
2236
// scheduler.IsNegativeScheduleIDOperator("not in") == true
2237
name: "NotIn",
2238
query: fmt.Sprintf("ScheduleId NOT IN ('%s')", sid1),
2239
wantIDs: []string{sid2},
2240
},
2241
{
2242
name: "IsNotNull",
2243
query: "ScheduleId IS NOT NULL",
2244
wantIDs: []string{sid1, sid2},
2245
},
2246
}
2247
2248
for _, tc := range tests {
2249
t.Run(tc.name, func(t *testing.T) {
2250
s.EventuallyWithT(func(c *assert.CollectT) {
2251
ids := listScheduleIDs(tc.query)
2252
require.Len(c, ids, len(tc.wantIDs))
2253
for _, want := range tc.wantIDs {
2254
require.Contains(c, ids, want)
2255
}
2256
}, 15*time.Second, 1*time.Second)
2257
})
2258
}
2259
}
2260
2261
func testScheduleInternalTaskQueue(t *testing.T, newContext contextFactory) {
2262
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
2263
errorMessageKeyword := "internal per-namespace task queue"
2264
2265
// Test CreateSchedule with internal task queue
2266
t.Run("CreateSchedule_PerNSWorkerTaskQueue", func(t *testing.T) {
2267
sid := "sched-test-internal-tq-create"
2268
schedule := &schedulepb.Schedule{
2269
Spec: &schedulepb.ScheduleSpec{
2270
Interval: []*schedulepb.IntervalSpec{
2271
{Interval: durationpb.New(1 * time.Hour)},
2272
},
2273
},
2274
Action: &schedulepb.ScheduleAction{
2275
Action: &schedulepb.ScheduleAction_StartWorkflow{
2276
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2277
WorkflowId: "wf-internal-tq",
2278
WorkflowType: &commonpb.WorkflowType{Name: "action"},
2279
TaskQueue: &taskqueuepb.TaskQueue{Name: primitives.PerNSWorkerTaskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2280
},
2281
},
2282
},
2283
}
2284
req := &workflowservice.CreateScheduleRequest{
2285
Namespace: s.Namespace().String(),
2286
ScheduleId: sid,
2287
Schedule: schedule,
2288
Identity: "test",
2289
RequestId: uuid.NewString(),
2290
}
2291
2292
ctx := newContext(s.Context())
2293
_, err := s.FrontendClient().CreateSchedule(ctx, req)
2294
require.Error(t, err)
2295
var invalidArgument *serviceerror.InvalidArgument
2296
require.ErrorAs(t, err, &invalidArgument)
2297
require.Contains(t, err.Error(), errorMessageKeyword)
2298
})
2299
2300
// Test UpdateSchedule with internal task queue
2301
t.Run("UpdateSchedule_PerNSWorkerTaskQueue", func(t *testing.T) {
2302
// First create a schedule with a valid task queue
2303
sid := "sched-test-internal-tq-update"
2304
schedule := &schedulepb.Schedule{
2305
Spec: &schedulepb.ScheduleSpec{
2306
Interval: []*schedulepb.IntervalSpec{
2307
{Interval: durationpb.New(1 * time.Hour)},
2308
},
2309
},
2310
Action: &schedulepb.ScheduleAction{
2311
Action: &schedulepb.ScheduleAction_StartWorkflow{
2312
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2313
WorkflowId: "wf-update-internal-tq",
2314
WorkflowType: &commonpb.WorkflowType{Name: "action"},
2315
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2316
},
2317
},
2318
},
2319
}
2320
req := &workflowservice.CreateScheduleRequest{
2321
Namespace: s.Namespace().String(),
2322
ScheduleId: sid,
2323
Schedule: schedule,
2324
Identity: "test",
2325
RequestId: uuid.NewString(),
2326
}
2327
2328
ctx := newContext(s.Context())
2329
_, err := s.FrontendClient().CreateSchedule(ctx, req)
2330
require.NoError(t, err)
2331
2332
// Now try to update with internal task queue
2333
schedule.Action.GetStartWorkflow().TaskQueue = &taskqueuepb.TaskQueue{
2334
Name: primitives.PerNSWorkerTaskQueue,
2335
Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
2336
}
2337
updateReq := &workflowservice.UpdateScheduleRequest{
2338
Namespace: s.Namespace().String(),
2339
ScheduleId: sid,
2340
Schedule: schedule,
2341
Identity: "test",
2342
RequestId: uuid.NewString(),
2343
}
2344
2345
_, err = s.FrontendClient().UpdateSchedule(ctx, updateReq)
2346
require.Error(t, err)
2347
var invalidArgument *serviceerror.InvalidArgument
2348
require.ErrorAs(t, err, &invalidArgument)
2349
require.Contains(t, err.Error(), errorMessageKeyword)
2350
})
2351
}
2352
2353
func testScheduledWorkflowDoubleReset(t *testing.T, newContext contextFactory, enableCHASMCallbacks bool) {
2354
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
2355
s.OverrideDynamicConfig(dynamicconfig.EnableCHASMCallbacks, enableCHASMCallbacks)
2356
2357
sid := "sched-test-double-reset"
2358
wid := "sched-test-double-reset-wf"
2359
wt := "sched-test-double-reset-wt"
2360
2361
s.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
2362
ch := workflow.GetSignalChannel(ctx, "complete")
2363
var signal any
2364
ch.Receive(ctx, &signal)
2365
return nil
2366
}, workflow.RegisterOptions{Name: wt})
2367
2368
ctx := newContext(s.Context())
2369
2370
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
2371
Namespace: s.Namespace().String(),
2372
ScheduleId: sid,
2373
Schedule: &schedulepb.Schedule{
2374
Spec: &schedulepb.ScheduleSpec{
2375
Interval: []*schedulepb.IntervalSpec{
2376
{Interval: durationpb.New(24 * time.Hour)},
2377
},
2378
},
2379
Action: &schedulepb.ScheduleAction{
2380
Action: &schedulepb.ScheduleAction_StartWorkflow{
2381
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2382
WorkflowId: wid,
2383
WorkflowType: &commonpb.WorkflowType{Name: wt},
2384
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2385
},
2386
},
2387
},
2388
},
2389
InitialPatch: &schedulepb.SchedulePatch{
2390
TriggerImmediately: &schedulepb.TriggerImmediatelyRequest{},
2391
},
2392
RequestId: uuid.NewString(),
2393
})
2394
s.NoError(err)
2395
2396
// Wait for scheduler to start the workflow and show it as RUNNING.
2397
listEntry := getScheduleEntryFromVisibility(s, sid, newContext, func(ent *schedulepb.ScheduleListEntry) bool {
2398
return len(ent.Info.RecentActions) >= 1 &&
2399
ent.Info.RecentActions[0].GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING
2400
})
2401
a1 := listEntry.Info.RecentActions[0]
2402
wfExec := &commonpb.WorkflowExecution{
2403
WorkflowId: a1.StartWorkflowResult.WorkflowId,
2404
RunId: a1.StartWorkflowResult.RunId,
2405
}
2406
2407
s.WaitForHistoryEvents(`
2408
1 WorkflowExecutionStarted
2409
2 WorkflowTaskScheduled
2410
3 WorkflowTaskStarted
2411
4 WorkflowTaskCompleted`,
2412
s.GetHistoryFunc(s.Namespace().String(), wfExec),
2413
5*time.Second,
2414
10*time.Millisecond,
2415
)
2416
2417
origDesc, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
2418
Namespace: s.Namespace().String(),
2419
Execution: wfExec,
2420
})
2421
s.NoError(err)
2422
var originalStartReqID string
2423
for reqID, info := range origDesc.GetWorkflowExtendedInfo().GetRequestIdInfos() {
2424
if info.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
2425
originalStartReqID = reqID
2426
break
2427
}
2428
}
2429
s.NotEmpty(originalStartReqID, "original run must have a request ID for WorkflowExecutionStarted")
2430
2431
resp1, err := s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
2432
Namespace: s.Namespace().String(),
2433
WorkflowExecution: wfExec,
2434
Reason: "double-reset-test-first",
2435
WorkflowTaskFinishEventId: 3,
2436
RequestId: uuid.NewString(),
2437
})
2438
s.NoError(err)
2439
resetRun1 := &commonpb.WorkflowExecution{
2440
WorkflowId: wfExec.WorkflowId,
2441
RunId: resp1.RunId,
2442
}
2443
2444
s.EventuallyWithT(func(col *assert.CollectT) {
2445
resetDesc, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
2446
Namespace: s.Namespace().String(),
2447
Execution: resetRun1,
2448
})
2449
require.NoError(col, err)
2450
var resetStartReqID string
2451
for reqID, info := range resetDesc.GetWorkflowExtendedInfo().GetRequestIdInfos() {
2452
if info.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
2453
resetStartReqID = reqID
2454
break
2455
}
2456
}
2457
require.Equal(col, originalStartReqID, resetStartReqID,
2458
"start request ID must be preserved across first reset")
2459
}, 10*time.Second, 100*time.Millisecond)
2460
2461
resp2, err := s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
2462
Namespace: s.Namespace().String(),
2463
WorkflowExecution: resetRun1,
2464
Reason: "double-reset-test-second",
2465
WorkflowTaskFinishEventId: 3,
2466
RequestId: uuid.NewString(),
2467
})
2468
s.NoError(err)
2469
resetRun2 := &commonpb.WorkflowExecution{
2470
WorkflowId: wfExec.WorkflowId,
2471
RunId: resp2.RunId,
2472
}
2473
2474
s.EventuallyWithT(func(col *assert.CollectT) {
2475
resetDesc, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
2476
Namespace: s.Namespace().String(),
2477
Execution: resetRun2,
2478
})
2479
require.NoError(col, err)
2480
var resetStartReqID string
2481
for reqID, info := range resetDesc.GetWorkflowExtendedInfo().GetRequestIdInfos() {
2482
if info.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
2483
resetStartReqID = reqID
2484
break
2485
}
2486
}
2487
require.Equal(col, originalStartReqID, resetStartReqID,
2488
"start request ID must be preserved across double reset")
2489
}, 10*time.Second, 100*time.Millisecond)
2490
2491
_, err = s.FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
2492
Namespace: s.Namespace().String(),
2493
WorkflowExecution: &commonpb.WorkflowExecution{
2494
WorkflowId: wfExec.WorkflowId,
2495
},
2496
SignalName: "complete",
2497
})
2498
s.NoError(err)
2499
2500
getScheduleEntryFromVisibility(s, sid, newContext, func(ent *schedulepb.ScheduleListEntry) bool {
2501
for _, action := range ent.Info.RecentActions {
2502
if action.GetStartWorkflowResult().GetRunId() == wfExec.RunId {
2503
return action.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
2504
}
2505
}
2506
return false
2507
})
2508
}
2509
2510
func testResetWithAdditionalCallback(t *testing.T, newContext contextFactory, enableCHASMCallbacks bool) {
2511
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
2512
s.OverrideDynamicConfig(dynamicconfig.EnableCHASMCallbacks, enableCHASMCallbacks)
2513
s.OverrideDynamicConfig(
2514
callback.AllowedAddresses,
2515
[]any{map[string]any{"Pattern": "*", "AllowInsecure": true}},
2516
)
2517
2518
sid := "sched-test-reset-extra-cb"
2519
wid := "sched-test-reset-extra-cb-wf"
2520
wt := "sched-test-reset-extra-cb-wt"
2521
2522
ch := &completionHandler{
2523
requestCh: make(chan *nexusrpc.CompletionRequest, 1),
2524
requestCompleteCh: make(chan error, 1),
2525
}
2526
defer func() {
2527
close(ch.requestCh)
2528
close(ch.requestCompleteCh)
2529
}()
2530
secondCallbackURL := func() string {
2531
hh := nexusrpc.NewCompletionHTTPHandler(nexusrpc.CompletionHandlerOptions{Handler: ch})
2532
srv := httptest.NewServer(hh)
2533
t.Cleanup(func() { srv.Close() })
2534
return srv.URL + "/callback"
2535
}()
2536
2537
s.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error {
2538
sigCh := workflow.GetSignalChannel(ctx, "complete")
2539
var signal any
2540
sigCh.Receive(ctx, &signal)
2541
return nil
2542
}, workflow.RegisterOptions{Name: wt})
2543
2544
ctx := newContext(s.Context())
2545
2546
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
2547
Namespace: s.Namespace().String(),
2548
ScheduleId: sid,
2549
Schedule: &schedulepb.Schedule{
2550
Spec: &schedulepb.ScheduleSpec{
2551
Interval: []*schedulepb.IntervalSpec{
2552
{Interval: durationpb.New(24 * time.Hour)},
2553
},
2554
},
2555
Action: &schedulepb.ScheduleAction{
2556
Action: &schedulepb.ScheduleAction_StartWorkflow{
2557
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2558
WorkflowId: wid,
2559
WorkflowType: &commonpb.WorkflowType{Name: wt},
2560
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2561
},
2562
},
2563
},
2564
},
2565
InitialPatch: &schedulepb.SchedulePatch{
2566
TriggerImmediately: &schedulepb.TriggerImmediatelyRequest{},
2567
},
2568
RequestId: uuid.NewString(),
2569
})
2570
s.NoError(err)
2571
2572
listEntry := getScheduleEntryFromVisibility(s, sid, newContext, func(ent *schedulepb.ScheduleListEntry) bool {
2573
return len(ent.Info.RecentActions) >= 1 &&
2574
ent.Info.RecentActions[0].GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING
2575
})
2576
a1 := listEntry.Info.RecentActions[0]
2577
wfExec := &commonpb.WorkflowExecution{
2578
WorkflowId: a1.StartWorkflowResult.WorkflowId,
2579
RunId: a1.StartWorkflowResult.RunId,
2580
}
2581
2582
s.WaitForHistoryEvents(`
2583
1 WorkflowExecutionStarted
2584
2 WorkflowTaskScheduled
2585
3 WorkflowTaskStarted
2586
4 WorkflowTaskCompleted`,
2587
s.GetHistoryFunc(s.Namespace().String(), wfExec),
2588
5*time.Second,
2589
10*time.Millisecond,
2590
)
2591
2592
attachRequestID := uuid.NewString()
2593
attachResp, err := s.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
2594
RequestId: attachRequestID,
2595
Namespace: s.Namespace().String(),
2596
WorkflowId: wfExec.WorkflowId,
2597
WorkflowType: &commonpb.WorkflowType{Name: wt},
2598
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2599
WorkflowIdConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
2600
OnConflictOptions: &workflowpb.OnConflictOptions{
2601
AttachRequestId: true,
2602
AttachCompletionCallbacks: true,
2603
},
2604
CompletionCallbacks: []*commonpb.Callback{
2605
{
2606
Variant: &commonpb.Callback_Nexus_{
2607
Nexus: &commonpb.Callback_Nexus{
2608
Url: secondCallbackURL,
2609
},
2610
},
2611
},
2612
},
2613
})
2614
s.NoError(err)
2615
s.False(attachResp.Started, "expected to attach to existing run, not start a new one")
2616
2617
s.WaitForHistoryEvents(`
2618
1 WorkflowExecutionStarted
2619
2 WorkflowTaskScheduled
2620
3 WorkflowTaskStarted
2621
4 WorkflowTaskCompleted
2622
5 WorkflowExecutionOptionsUpdated`,
2623
s.GetHistoryFunc(s.Namespace().String(), wfExec),
2624
5*time.Second,
2625
10*time.Millisecond,
2626
)
2627
2628
resetResp, err := s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
2629
Namespace: s.Namespace().String(),
2630
WorkflowExecution: wfExec,
2631
Reason: "reset-with-additional-callback-test",
2632
WorkflowTaskFinishEventId: 3,
2633
RequestId: uuid.NewString(),
2634
})
2635
s.NoError(err)
2636
resetRun := &commonpb.WorkflowExecution{
2637
WorkflowId: wfExec.WorkflowId,
2638
RunId: resetResp.RunId,
2639
}
2640
2641
var startRequestID string
2642
s.EventuallyWithT(func(col *assert.CollectT) {
2643
descResp, err := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
2644
Namespace: s.Namespace().String(),
2645
Execution: resetRun,
2646
})
2647
require.NoError(col, err)
2648
require.Len(col, descResp.Callbacks, 2)
2649
reqIDs := descResp.GetWorkflowExtendedInfo().GetRequestIdInfos()
2650
attachInfo, ok := reqIDs[attachRequestID]
2651
require.True(col, ok, "attachRequestId not found in RequestIdInfos")
2652
require.Equal(col, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED, attachInfo.GetEventType())
2653
for reqID, info := range reqIDs {
2654
if info.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
2655
startRequestID = reqID
2656
break
2657
}
2658
}
2659
require.NotEmpty(col, startRequestID, "no request ID found for WorkflowExecutionStarted")
2660
require.NotEqual(col, startRequestID, attachRequestID,
2661
"schedule callback and manually-attached callback must have different request IDs")
2662
}, 10*time.Second, 100*time.Millisecond)
2663
2664
_, err = s.FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
2665
Namespace: s.Namespace().String(),
2666
WorkflowExecution: &commonpb.WorkflowExecution{
2667
WorkflowId: wfExec.WorkflowId,
2668
},
2669
SignalName: "complete",
2670
})
2671
s.NoError(err)
2672
2673
getScheduleEntryFromVisibility(s, sid, newContext, func(ent *schedulepb.ScheduleListEntry) bool {
2674
for _, action := range ent.Info.RecentActions {
2675
if action.GetStartWorkflowResult().GetRunId() == wfExec.RunId {
2676
return action.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
2677
}
2678
}
2679
return false
2680
})
2681
2682
select {
2683
case completion := <-ch.requestCh:
2684
s.Equal(nexus.OperationStateSucceeded, completion.State)
2685
ch.requestCompleteCh <- nil
2686
case <-time.After(10 * time.Second):
2687
s.Fail("timeout waiting for second callback to be delivered")
2688
}
2689
}
2690
2691
// testCreatesWorkflowSentinel tests that creating a CHASM schedule also starts a
2692
// dummy workflow to reserve the schedule ID in the V1 workflow ID-space.
2693
func testCreatesWorkflowSentinel(t *testing.T, newContext contextFactory) {
2694
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
2695
2696
sid := testcore.RandomizeStr("sid")
2697
wid := testcore.RandomizeStr("wid")
2698
wt := testcore.RandomizeStr("wt")
2699
2700
schedule := &schedulepb.Schedule{
2701
Spec: &schedulepb.ScheduleSpec{
2702
Interval: []*schedulepb.IntervalSpec{
2703
{Interval: durationpb.New(1 * time.Hour)},
2704
},
2705
},
2706
Action: &schedulepb.ScheduleAction{
2707
Action: &schedulepb.ScheduleAction_StartWorkflow{
2708
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2709
WorkflowId: wid,
2710
WorkflowType: &commonpb.WorkflowType{Name: wt},
2711
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2712
},
2713
},
2714
},
2715
}
2716
2717
ctx := newContext(s.Context())
2718
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
2719
Namespace: s.Namespace().String(),
2720
ScheduleId: sid,
2721
Schedule: schedule,
2722
Identity: testcore.RandomizeStr("identity"),
2723
RequestId: testcore.RandomizeStr("request-id"),
2724
})
2725
s.NoError(err)
2726
2727
// Verify the dummy workflow was created to reserve the V1 workflow ID.
2728
sentinelWfID := scheduler.WorkflowIDPrefix + sid
2729
var descResp *workflowservice.DescribeWorkflowExecutionResponse
2730
s.Eventually(func() bool {
2731
descResp, err = s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
2732
Namespace: s.Namespace().String(),
2733
Execution: &commonpb.WorkflowExecution{WorkflowId: sentinelWfID},
2734
})
2735
return err == nil
2736
}, 15*time.Second, 500*time.Millisecond, "dummy sentinel workflow should exist")
2737
s.Equal(dummy.DummyWFTypeName, descResp.WorkflowExecutionInfo.Type.Name)
2738
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, descResp.WorkflowExecutionInfo.Status)
2739
2740
// Verify visibility shows exactly one schedule (not the dummy workflow).
2741
getScheduleEntryFromVisibility(s, sid, newContext, nil)
2742
listResp, err := s.FrontendClient().ListSchedules(ctx, &workflowservice.ListSchedulesRequest{
2743
Namespace: s.Namespace().String(),
2744
MaximumPageSize: 5,
2745
})
2746
s.NoError(err)
2747
s.Len(listResp.Schedules, 1)
2748
2749
countResp, err := s.FrontendClient().CountSchedules(ctx, &workflowservice.CountSchedulesRequest{
2750
Namespace: s.Namespace().String(),
2751
})
2752
s.NoError(err)
2753
s.Equal(int64(1), countResp.Count)
2754
}
2755
2756
func testStateSizeBytesReported(t *testing.T, newContext contextFactory) {
2757
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
2758
2759
sid := testcore.RandomizeStr("sched-state-size")
2760
wid := testcore.RandomizeStr("sched-state-size-wf")
2761
wt := testcore.RandomizeStr("sched-state-size-wt")
2762
2763
schedule := &schedulepb.Schedule{
2764
Spec: &schedulepb.ScheduleSpec{
2765
Interval: []*schedulepb.IntervalSpec{
2766
{Interval: durationpb.New(1 * time.Hour)},
2767
},
2768
},
2769
Action: &schedulepb.ScheduleAction{
2770
Action: &schedulepb.ScheduleAction_StartWorkflow{
2771
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2772
WorkflowId: wid,
2773
WorkflowType: &commonpb.WorkflowType{Name: wt},
2774
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2775
},
2776
},
2777
},
2778
State: &schedulepb.ScheduleState{Paused: true},
2779
}
2780
2781
ctx := newContext(s.Context())
2782
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
2783
Namespace: s.Namespace().String(),
2784
ScheduleId: sid,
2785
Schedule: schedule,
2786
Identity: "test",
2787
RequestId: uuid.NewString(),
2788
})
2789
s.NoError(err)
2790
2791
desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
2792
Namespace: s.Namespace().String(),
2793
ScheduleId: sid,
2794
})
2795
s.NoError(err)
2796
s.Positive(desc.GetInfo().GetStateSizeBytes(), "Describe should report a non-zero StateSizeBytes")
2797
}
2798
2799
// testCreatesCHASMSentinel tests that creating a V1 schedule also creates a
2800
// CHASM sentinel to reserve the schedule ID in the CHASM execution space.
2801
func testCreatesCHASMSentinel(t *testing.T, newContext contextFactory) {
2802
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
2803
2804
sid := testcore.RandomizeStr("sid")
2805
wid := testcore.RandomizeStr("wid")
2806
wt := testcore.RandomizeStr("wt")
2807
2808
schedule := &schedulepb.Schedule{
2809
Spec: &schedulepb.ScheduleSpec{
2810
Interval: []*schedulepb.IntervalSpec{
2811
{Interval: durationpb.New(1 * time.Hour)},
2812
},
2813
},
2814
Action: &schedulepb.ScheduleAction{
2815
Action: &schedulepb.ScheduleAction_StartWorkflow{
2816
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2817
WorkflowId: wid,
2818
WorkflowType: &commonpb.WorkflowType{Name: wt},
2819
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2820
},
2821
},
2822
},
2823
}
2824
2825
ctx := newContext(s.Context())
2826
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
2827
Namespace: s.Namespace().String(),
2828
ScheduleId: sid,
2829
Schedule: schedule,
2830
Identity: testcore.RandomizeStr("identity"),
2831
RequestId: testcore.RandomizeStr("request-id"),
2832
})
2833
s.NoError(err)
2834
2835
// Verify a CHASM sentinel was created to reserve the schedule ID.
2836
// DescribeSchedule should return NotFound, as well as CreateSentinel
2837
nsID := s.NamespaceID().String()
2838
s.Eventually(func() bool {
2839
_, descErr := s.GetTestCluster().SchedulerClient().DescribeSchedule(
2840
ctx,
2841
&schedulerpb.DescribeScheduleRequest{
2842
NamespaceId: nsID,
2843
FrontendRequest: &workflowservice.DescribeScheduleRequest{Namespace: s.Namespace().String(), ScheduleId: sid},
2844
},
2845
)
2846
var notFoundErr *serviceerror.NotFound
2847
if !errors.As(descErr, ¬FoundErr) {
2848
return false
2849
}
2850
2851
// A CHASM CreateSchedule should also fail with NotFound because
2852
// the sentinel blocks it.
2853
_, createErr := s.GetTestCluster().SchedulerClient().CreateSchedule(
2854
ctx,
2855
&schedulerpb.CreateScheduleRequest{
2856
NamespaceId: nsID,
2857
FrontendRequest: &workflowservice.CreateScheduleRequest{
2858
Namespace: s.Namespace().String(),
2859
ScheduleId: sid,
2860
RequestId: testcore.RandomizeStr("test-sentinel-check"),
2861
Schedule: schedule,
2862
},
2863
},
2864
)
2865
return errors.As(createErr, ¬FoundErr)
2866
}, 15*time.Second, 500*time.Millisecond, "CHASM sentinel should exist for V1 schedule")
2867
2868
// Verify visibility shows exactly one schedule (not the sentinel).
2869
getScheduleEntryFromVisibility(s, sid, newContext, nil)
2870
listResp, err := s.FrontendClient().ListSchedules(ctx, &workflowservice.ListSchedulesRequest{
2871
Namespace: s.Namespace().String(),
2872
MaximumPageSize: 5,
2873
})
2874
s.NoError(err)
2875
s.Len(listResp.Schedules, 1)
2876
2877
countResp, err := s.FrontendClient().CountSchedules(ctx, &workflowservice.CountSchedulesRequest{
2878
Namespace: s.Namespace().String(),
2879
})
2880
s.NoError(err)
2881
s.Equal(int64(1), countResp.Count)
2882
}
2883
2884
// testSkipsWorkflowSentinelWhenDisabled asserts that a CHASM CreateSchedule
2885
// does not start the dummy V1 workflow when EnableCHASMSchedulerSentinels is off.
2886
func testSkipsWorkflowSentinelWhenDisabled(t *testing.T, newContext contextFactory) {
2887
s := newScheduleEnv(t, append(scheduleCommonOpts(t),
2888
testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerSentinels, false),
2889
)...)
2890
2891
sid := testcore.RandomizeStr("sid")
2892
wid := testcore.RandomizeStr("wid")
2893
wt := testcore.RandomizeStr("wt")
2894
2895
schedule := &schedulepb.Schedule{
2896
Spec: &schedulepb.ScheduleSpec{
2897
Interval: []*schedulepb.IntervalSpec{
2898
{Interval: durationpb.New(1 * time.Hour)},
2899
},
2900
},
2901
Action: &schedulepb.ScheduleAction{
2902
Action: &schedulepb.ScheduleAction_StartWorkflow{
2903
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2904
WorkflowId: wid,
2905
WorkflowType: &commonpb.WorkflowType{Name: wt},
2906
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2907
},
2908
},
2909
},
2910
}
2911
2912
ctx := newContext(s.Context())
2913
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
2914
Namespace: s.Namespace().String(),
2915
ScheduleId: sid,
2916
Schedule: schedule,
2917
Identity: testcore.RandomizeStr("identity"),
2918
RequestId: testcore.RandomizeStr("request-id"),
2919
})
2920
s.NoError(err)
2921
2922
// The dummy V1 workflow that reserves the schedule ID is gated on the
2923
// sentinel flag, so it must not exist.
2924
sentinelWfID := scheduler.WorkflowIDPrefix + sid
2925
_, descErr := s.FrontendClient().DescribeWorkflowExecution(ctx, &workflowservice.DescribeWorkflowExecutionRequest{
2926
Namespace: s.Namespace().String(),
2927
Execution: &commonpb.WorkflowExecution{WorkflowId: sentinelWfID},
2928
})
2929
var notFoundErr *serviceerror.NotFound
2930
s.ErrorAs(descErr, ¬FoundErr, "no dummy sentinel workflow should be created when sentinels are disabled")
2931
}
2932
2933
// testSkipsCHASMSentinelWhenDisabled asserts that a V1 CreateSchedule does not
2934
// create a CHASM sentinel when EnableCHASMSchedulerSentinels is off.
2935
func testSkipsCHASMSentinelWhenDisabled(t *testing.T, newContext contextFactory) {
2936
s := newScheduleEnv(t, append(scheduleCommonOpts(t),
2937
testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerSentinels, false),
2938
)...)
2939
2940
sid := testcore.RandomizeStr("sid")
2941
wid := testcore.RandomizeStr("wid")
2942
wt := testcore.RandomizeStr("wt")
2943
2944
schedule := &schedulepb.Schedule{
2945
Spec: &schedulepb.ScheduleSpec{
2946
Interval: []*schedulepb.IntervalSpec{
2947
{Interval: durationpb.New(1 * time.Hour)},
2948
},
2949
},
2950
Action: &schedulepb.ScheduleAction{
2951
Action: &schedulepb.ScheduleAction_StartWorkflow{
2952
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
2953
WorkflowId: wid,
2954
WorkflowType: &commonpb.WorkflowType{Name: wt},
2955
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
2956
},
2957
},
2958
},
2959
}
2960
2961
ctx := newContext(s.Context())
2962
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
2963
Namespace: s.Namespace().String(),
2964
ScheduleId: sid,
2965
Schedule: schedule,
2966
Identity: testcore.RandomizeStr("identity"),
2967
RequestId: testcore.RandomizeStr("request-id"),
2968
})
2969
s.NoError(err)
2970
2971
// With no CHASM sentinel reserving the ID, a CHASM CreateSchedule for the
2972
// same ID must not be blocked by the NotFound (sentinel) signal.
2973
nsID := s.NamespaceID().String()
2974
_, createErr := s.GetTestCluster().SchedulerClient().CreateSchedule(
2975
ctx,
2976
&schedulerpb.CreateScheduleRequest{
2977
NamespaceId: nsID,
2978
FrontendRequest: &workflowservice.CreateScheduleRequest{
2979
Namespace: s.Namespace().String(),
2980
ScheduleId: sid,
2981
RequestId: testcore.RandomizeStr("test-no-sentinel"),
2982
Schedule: schedule,
2983
},
2984
},
2985
)
2986
s.NoError(createErr, "no CHASM sentinel should block CreateSchedule when sentinels are disabled, got: %v", createErr)
2987
}
2988
2989
func testCreateScheduleAlreadyExists(t *testing.T, newContext contextFactory) {
2990
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
2991
2992
sid := "sched-test-already-exists"
2993
2994
schedule := &schedulepb.Schedule{
2995
Spec: &schedulepb.ScheduleSpec{
2996
Interval: []*schedulepb.IntervalSpec{
2997
{Interval: durationpb.New(1 * time.Hour)},
2998
},
2999
},
3000
Action: &schedulepb.ScheduleAction{
3001
Action: &schedulepb.ScheduleAction_StartWorkflow{
3002
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3003
WorkflowId: "wf-already-exists",
3004
WorkflowType: &commonpb.WorkflowType{Name: "action"},
3005
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3006
},
3007
},
3008
},
3009
}
3010
req := &workflowservice.CreateScheduleRequest{
3011
Namespace: s.Namespace().String(),
3012
ScheduleId: sid,
3013
Schedule: schedule,
3014
Identity: "test",
3015
RequestId: uuid.NewString(),
3016
}
3017
3018
ctx := newContext(s.Context())
3019
_, err := s.FrontendClient().CreateSchedule(ctx, req)
3020
s.NoError(err)
3021
3022
// Try to create again with a different request ID - should fail with AlreadyExists
3023
req.RequestId = uuid.NewString()
3024
_, err = s.FrontendClient().CreateSchedule(ctx, req)
3025
s.Error(err)
3026
3027
var alreadyStarted *serviceerror.WorkflowExecutionAlreadyStarted
3028
s.ErrorAs(err, &alreadyStarted)
3029
s.Contains(err.Error(), sid)
3030
}
3031
3032
// CreateSchedule is special-cased in the SDKs to translate
3033
// serviceerror.WorkflowExecutionAlreadyStarted into
3034
// temporal.ErrScheduleAlreadyRunning. This tests the SDK's behavior E2E against
3035
// the handler. A similar test exists in the features repository.
3036
func testCreateScheduleDuplicateSdkError(t *testing.T, useCHASM bool) {
3037
opts := scheduleCommonOpts(t)
3038
if useCHASM {
3039
opts = append(opts, testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true))
3040
}
3041
s := newScheduleEnv(t, opts...)
3042
3043
sid := "sched-test-duplicate-sdk-" + uuid.NewString()[:8]
3044
schedOpts := sdkclient.ScheduleOptions{
3045
ID: sid,
3046
Spec: sdkclient.ScheduleSpec{},
3047
Action: &sdkclient.ScheduleWorkflowAction{
3048
ID: "wf-" + sid,
3049
Workflow: "noop",
3050
TaskQueue: s.WorkerTaskQueue(),
3051
},
3052
Paused: true,
3053
}
3054
3055
ctx := s.Context()
3056
handle, err := s.SdkClient().ScheduleClient().Create(ctx, schedOpts)
3057
s.NoError(err)
3058
defer func() { _ = handle.Delete(context.Background()) }()
3059
3060
_, err = s.SdkClient().ScheduleClient().Create(ctx, schedOpts)
3061
s.ErrorIs(err, temporal.ErrScheduleAlreadyRunning)
3062
}
3063
3064
func testPatchRejectsExcessBackfillers(t *testing.T, newContext contextFactory) {
3065
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
3066
sid := "sched-test-too-many-backfillers"
3067
wt := "sched-test-too-many-backfillers-wt"
3068
3069
schedule := &schedulepb.Schedule{
3070
Spec: &schedulepb.ScheduleSpec{
3071
Interval: []*schedulepb.IntervalSpec{
3072
{Interval: durationpb.New(1 * time.Hour)},
3073
},
3074
},
3075
Action: &schedulepb.ScheduleAction{
3076
Action: &schedulepb.ScheduleAction_StartWorkflow{
3077
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3078
WorkflowId: "wf-too-many-backfillers",
3079
WorkflowType: &commonpb.WorkflowType{Name: wt},
3080
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3081
},
3082
},
3083
},
3084
State: &schedulepb.ScheduleState{Paused: true},
3085
}
3086
3087
ctx := newContext(s.Context())
3088
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
3089
Namespace: s.Namespace().String(),
3090
ScheduleId: sid,
3091
Schedule: schedule,
3092
Identity: "test",
3093
RequestId: uuid.NewString(),
3094
})
3095
s.NoError(err)
3096
3097
// Patch with 50 backfill requests at a time until we reach the limit of 100.
3098
now := time.Now()
3099
for i := 0; i < 100; i += 50 {
3100
backfills := make([]*schedulepb.BackfillRequest, 50)
3101
for j := range backfills {
3102
backfills[j] = &schedulepb.BackfillRequest{
3103
StartTime: timestamppb.New(now),
3104
EndTime: timestamppb.New(now.Add(time.Minute)),
3105
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
3106
}
3107
}
3108
_, err = s.FrontendClient().PatchSchedule(ctx, &workflowservice.PatchScheduleRequest{
3109
Namespace: s.Namespace().String(),
3110
ScheduleId: sid,
3111
Patch: &schedulepb.SchedulePatch{
3112
BackfillRequest: backfills,
3113
},
3114
Identity: "test",
3115
RequestId: uuid.NewString(),
3116
})
3117
s.NoError(err)
3118
}
3119
3120
// The next patch should be rejected.
3121
_, err = s.FrontendClient().PatchSchedule(ctx, &workflowservice.PatchScheduleRequest{
3122
Namespace: s.Namespace().String(),
3123
ScheduleId: sid,
3124
Patch: &schedulepb.SchedulePatch{
3125
BackfillRequest: []*schedulepb.BackfillRequest{
3126
{
3127
StartTime: timestamppb.New(now),
3128
EndTime: timestamppb.New(now.Add(time.Minute)),
3129
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
3130
},
3131
},
3132
},
3133
Identity: "test",
3134
RequestId: uuid.NewString(),
3135
})
3136
s.Error(err)
3137
var failedPrecondition *serviceerror.FailedPrecondition
3138
s.ErrorAs(err, &failedPrecondition)
3139
s.Contains(err.Error(), "too many concurrent backfillers")
3140
}
3141
3142
func testMigrationCallbackAttach(t *testing.T, newContext contextFactory) {
3143
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
3144
3145
sid := testcore.RandomizeStr("sid")
3146
wid := testcore.RandomizeStr("wid")
3147
wt := testcore.RandomizeStr("wt")
3148
3149
resumeSignal := "resume"
3150
s.SdkWorker().RegisterWorkflowWithOptions(
3151
func(ctx workflow.Context) error {
3152
workflow.GetSignalChannel(ctx, resumeSignal).Receive(ctx, nil)
3153
return nil
3154
},
3155
workflow.RegisterOptions{Name: wt},
3156
)
3157
3158
ctx := newContext(s.Context())
3159
startResp, err := s.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
3160
Namespace: s.Namespace().String(),
3161
WorkflowId: wid,
3162
WorkflowType: &commonpb.WorkflowType{Name: wt},
3163
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3164
Identity: testcore.RandomizeStr("identity"),
3165
RequestId: testcore.RandomizeStr("request-id"),
3166
})
3167
s.NoError(err)
3168
3169
schedule := &schedulepb.Schedule{
3170
Spec: &schedulepb.ScheduleSpec{
3171
Interval: []*schedulepb.IntervalSpec{
3172
{Interval: durationpb.New(24 * time.Hour)},
3173
},
3174
},
3175
Action: &schedulepb.ScheduleAction{
3176
Action: &schedulepb.ScheduleAction_StartWorkflow{
3177
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3178
WorkflowId: wid,
3179
WorkflowType: &commonpb.WorkflowType{Name: wt},
3180
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3181
},
3182
},
3183
},
3184
}
3185
3186
now := time.Now().UTC()
3187
nsID := s.NamespaceID().String()
3188
3189
migrationState := &schedulerpb.SchedulerMigrationState{
3190
SchedulerState: &schedulerpb.SchedulerState{
3191
Namespace: s.Namespace().String(),
3192
NamespaceId: nsID,
3193
ScheduleId: sid,
3194
Schedule: schedule,
3195
Info: &schedulepb.ScheduleInfo{},
3196
ConflictToken: 1,
3197
},
3198
GeneratorState: &schedulerpb.GeneratorState{},
3199
InvokerState: &schedulerpb.InvokerState{
3200
BufferedStarts: []*schedulespb.BufferedStart{
3201
{
3202
NominalTime: timestamppb.New(now),
3203
ActualTime: timestamppb.New(now),
3204
StartTime: timestamppb.New(now),
3205
WorkflowId: wid,
3206
RunId: startResp.RunId,
3207
RequestId: uuid.NewString(),
3208
Attempt: 1,
3209
HasCallback: false,
3210
},
3211
},
3212
},
3213
}
3214
_, err = s.GetTestCluster().SchedulerClient().CreateFromMigrationState(
3215
ctx,
3216
&schedulerpb.CreateFromMigrationStateRequest{
3217
NamespaceId: nsID,
3218
State: migrationState,
3219
},
3220
)
3221
s.NoError(err)
3222
3223
s.Eventually(func() bool {
3224
descResp, err := s.GetTestCluster().SchedulerClient().DescribeSchedule(
3225
ctx,
3226
&schedulerpb.DescribeScheduleRequest{
3227
NamespaceId: nsID,
3228
FrontendRequest: &workflowservice.DescribeScheduleRequest{Namespace: s.Namespace().String(), ScheduleId: sid},
3229
},
3230
)
3231
if err != nil {
3232
return false
3233
}
3234
running := descResp.GetFrontendResponse().GetInfo().GetRunningWorkflows()
3235
return len(running) > 0 && running[0].WorkflowId == wid
3236
}, 15*time.Second, 500*time.Millisecond, "CHASM scheduler should show running workflow")
3237
3238
_, err = s.FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
3239
Namespace: s.Namespace().String(),
3240
WorkflowExecution: &commonpb.WorkflowExecution{
3241
WorkflowId: wid,
3242
RunId: startResp.RunId,
3243
},
3244
SignalName: resumeSignal,
3245
})
3246
s.NoError(err)
3247
3248
s.Eventually(func() bool {
3249
descResp, err := s.GetTestCluster().SchedulerClient().DescribeSchedule(
3250
ctx,
3251
&schedulerpb.DescribeScheduleRequest{
3252
NamespaceId: nsID,
3253
FrontendRequest: &workflowservice.DescribeScheduleRequest{Namespace: s.Namespace().String(), ScheduleId: sid},
3254
},
3255
)
3256
if err != nil {
3257
return false
3258
}
3259
recent := descResp.GetFrontendResponse().GetInfo().GetRecentActions()
3260
for _, action := range recent {
3261
if action.GetStartWorkflowResult().GetWorkflowId() == wid &&
3262
action.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED {
3263
return true
3264
}
3265
}
3266
return false
3267
}, 15*time.Second, 500*time.Millisecond, "CHASM scheduler should reflect workflow completion")
3268
}
3269
3270
// testCHASMCanListV1Schedules tests that a schedule created in the V1 stack
3271
// will also be visible in the V2 stack.
3272
func testCHASMCanListV1Schedules(t *testing.T, newContext contextFactory) {
3273
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
3274
3275
sid := "schedule-created-on-v1"
3276
schedule := &schedulepb.Schedule{
3277
Spec: &schedulepb.ScheduleSpec{
3278
Interval: []*schedulepb.IntervalSpec{
3279
{Interval: durationpb.New(3 * time.Second)},
3280
},
3281
},
3282
Action: &schedulepb.ScheduleAction{
3283
Action: &schedulepb.ScheduleAction_StartWorkflow{
3284
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3285
WorkflowId: "wf-",
3286
WorkflowType: &commonpb.WorkflowType{Name: "action"},
3287
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3288
},
3289
},
3290
},
3291
}
3292
req := &workflowservice.CreateScheduleRequest{
3293
Namespace: s.Namespace().String(),
3294
ScheduleId: sid,
3295
Schedule: schedule,
3296
Identity: "test",
3297
RequestId: uuid.NewString(),
3298
}
3299
3300
// Create on V1 stack.
3301
_, err := s.FrontendClient().CreateSchedule(newContext(s.Context()), req)
3302
s.NoError(err)
3303
3304
// Pause so that `FutureActionTimes` doesn't change between calls.
3305
_, err = s.FrontendClient().PatchSchedule(newContext(s.Context()), &workflowservice.PatchScheduleRequest{
3306
Namespace: s.Namespace().String(),
3307
ScheduleId: sid,
3308
Patch: &schedulepb.SchedulePatch{
3309
Pause: "halt",
3310
},
3311
Identity: "test",
3312
RequestId: uuid.NewString(),
3313
})
3314
s.NoError(err)
3315
3316
// Sanity test, list with V1 handler.
3317
v1Entry := getScheduleEntryFromVisibility(s, sid, newContext, func(sle *schedulepb.ScheduleListEntry) bool {
3318
return sle.GetInfo().Paused
3319
})
3320
s.NotNil(v1Entry.GetInfo())
3321
3322
// Count with V1 handler.
3323
v1CountResp, err := s.FrontendClient().CountSchedules(newContext(s.Context()), &workflowservice.CountSchedulesRequest{
3324
Namespace: s.Namespace().String(),
3325
})
3326
s.NoError(err)
3327
s.GreaterOrEqual(v1CountResp.Count, int64(1), "Expected at least 1 schedule with V1 handler")
3328
3329
// Flip on CHASM experiment and make sure we can still list.
3330
chasmEntry := getScheduleEntryFromVisibility(s, sid, chasmContextFactory, nil)
3331
s.NotNil(chasmEntry.GetInfo())
3332
s.ProtoEqual(chasmEntry.GetInfo(), v1Entry.GetInfo())
3333
3334
// Count with CHASM handler and verify it matches V1 count.
3335
chasmCountResp, err := s.FrontendClient().CountSchedules(chasmContextFactory(s.Context()), &workflowservice.CountSchedulesRequest{
3336
Namespace: s.Namespace().String(),
3337
})
3338
s.NoError(err)
3339
s.Equal(v1CountResp.Count, chasmCountResp.Count, "CHASM and V1 counts should match")
3340
}
3341
3342
// testRefresh applies to V1 scheduler only; V2 does not support/need manual refresh.
3343
func testRefresh(t *testing.T, newContext contextFactory) {
3344
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
3345
3346
sid := "sched-test-refresh"
3347
wid := "sched-test-refresh-wf"
3348
wt := "sched-test-refresh-wt"
3349
3350
// Phase is computed so the first tick lands ~10s after this point. Under
3351
// parallel load CreateSchedule can take several seconds; if the first
3352
// tick passes before the server materializes the schedule we'd wait a
3353
// full 30s for the next one, exceeding the Eventually budget below.
3354
phaseOffset := 10 * time.Second
3355
schedule := &schedulepb.Schedule{
3356
Spec: &schedulepb.ScheduleSpec{
3357
Interval: []*schedulepb.IntervalSpec{
3358
{
3359
Interval: durationpb.New(30 * time.Second),
3360
Phase: durationpb.New(time.Duration((time.Now().Unix()+int64(phaseOffset/time.Second))%30) * time.Second),
3361
},
3362
},
3363
},
3364
Action: &schedulepb.ScheduleAction{
3365
Action: &schedulepb.ScheduleAction_StartWorkflow{
3366
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3367
WorkflowId: wid,
3368
WorkflowType: &commonpb.WorkflowType{Name: wt},
3369
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3370
WorkflowExecutionTimeout: durationpb.New(3 * time.Second),
3371
},
3372
},
3373
},
3374
}
3375
req := &workflowservice.CreateScheduleRequest{
3376
Namespace: s.Namespace().String(),
3377
ScheduleId: sid,
3378
Schedule: schedule,
3379
Identity: "test",
3380
RequestId: uuid.NewString(),
3381
}
3382
3383
var runs int32
3384
workflowFn := func(ctx workflow.Context) error {
3385
workflow.SideEffect(ctx, func(ctx workflow.Context) any {
3386
atomic.AddInt32(&runs, 1)
3387
return 0
3388
})
3389
s.NoError(workflow.Sleep(ctx, 10*time.Second)) // longer than execution timeout
3390
return nil
3391
}
3392
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
3393
3394
_, err := s.FrontendClient().CreateSchedule(newContext(s.Context()), req)
3395
s.NoError(err)
3396
3397
s.Eventually(func() bool { return atomic.LoadInt32(&runs) == 1 }, 20*time.Second, 200*time.Millisecond)
3398
3399
// workflow has started but is now sleeping. it will timeout in 2 seconds.
3400
3401
describeResp, err := s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
3402
Namespace: s.Namespace().String(),
3403
ScheduleId: sid,
3404
})
3405
s.NoError(err)
3406
s.Len(describeResp.Info.RunningWorkflows, 1)
3407
3408
events1 := s.GetHistory(s.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: scheduler.WorkflowIDPrefix + sid})
3409
expectedHistory := `
3410
1 WorkflowExecutionStarted
3411
2 WorkflowTaskScheduled
3412
3 WorkflowTaskStarted
3413
4 WorkflowTaskCompleted
3414
5 MarkerRecorded
3415
6 MarkerRecorded
3416
7 UpsertWorkflowSearchAttributes
3417
8 TimerStarted
3418
9 TimerFired
3419
10 WorkflowTaskScheduled
3420
11 WorkflowTaskStarted
3421
12 WorkflowTaskCompleted
3422
13 MarkerRecorded
3423
14 MarkerRecorded
3424
15 WorkflowPropertiesModified
3425
16 TimerStarted`
3426
3427
s.EqualHistoryEvents(expectedHistory, events1)
3428
3429
time.Sleep(4 * time.Second) //nolint:forbidigo
3430
// now it has timed out, but the scheduler hasn't noticed yet. we can prove it by checking
3431
// its history.
3432
3433
events2 := s.GetHistory(s.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: scheduler.WorkflowIDPrefix + sid})
3434
s.EqualHistoryEvents(expectedHistory, events2)
3435
3436
// when we describe we'll force a refresh and see it timed out
3437
describeResp, err = s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
3438
Namespace: s.Namespace().String(),
3439
ScheduleId: sid,
3440
})
3441
s.NoError(err)
3442
s.Empty(describeResp.Info.RunningWorkflows)
3443
3444
// check scheduler has gotten the refresh and done some stuff. signal is sent without waiting so we need to wait.
3445
s.Eventually(func() bool {
3446
events3 := s.GetHistory(s.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: scheduler.WorkflowIDPrefix + sid})
3447
return len(events3) > len(events2)
3448
}, 5*time.Second, 100*time.Millisecond)
3449
}
3450
3451
// testListBeforeRun only applies to V1, as V2 scheduler does not involve the
3452
// per-NS worker or workflow.
3453
func testListBeforeRun(t *testing.T, newContext contextFactory) {
3454
s := newScheduleEnv(t, append(scheduleCommonOpts(t),
3455
testcore.WithDynamicConfig(dynamicconfig.WorkerPerNamespaceWorkerCount, 0),
3456
)...)
3457
3458
sid := "sched-test-list-before-run"
3459
wid := "sched-test-list-before-run-wf"
3460
wt := "sched-test-list-before-run-wt"
3461
3462
schedule := &schedulepb.Schedule{
3463
Spec: &schedulepb.ScheduleSpec{
3464
Interval: []*schedulepb.IntervalSpec{
3465
{Interval: durationpb.New(3 * time.Second)},
3466
},
3467
},
3468
Action: &schedulepb.ScheduleAction{
3469
Action: &schedulepb.ScheduleAction_StartWorkflow{
3470
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3471
WorkflowId: wid,
3472
WorkflowType: &commonpb.WorkflowType{Name: wt},
3473
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3474
},
3475
},
3476
},
3477
}
3478
req := &workflowservice.CreateScheduleRequest{
3479
Namespace: s.Namespace().String(),
3480
ScheduleId: sid,
3481
Schedule: schedule,
3482
Identity: "test",
3483
RequestId: uuid.NewString(),
3484
}
3485
3486
startTime := time.Now()
3487
3488
_, err := s.FrontendClient().CreateSchedule(newContext(s.Context()), req)
3489
s.NoError(err)
3490
3491
entry := getScheduleEntryFromVisibility(s, sid, newContext, nil)
3492
s.NotNil(entry.Info)
3493
s.ProtoEqual(schedule.Spec, entry.Info.Spec)
3494
s.Equal(wt, entry.Info.WorkflowType.Name)
3495
s.False(entry.Info.Paused)
3496
s.Greater(len(entry.Info.FutureActionTimes), 1)
3497
s.True(entry.Info.FutureActionTimes[0].AsTime().After(startTime))
3498
}
3499
3500
// testRateLimit applies only to V1, as V2 scheduler does not impose its own rate limiting.
3501
func testRateLimit(t *testing.T, newContext contextFactory) {
3502
s := newScheduleEnv(t, append(scheduleCommonOpts(t),
3503
testcore.WithDynamicConfig(dynamicconfig.SchedulerNamespaceStartWorkflowRPS, 1.0),
3504
)...)
3505
3506
sid := "sched-test-rate-limit-%d"
3507
wid := "sched-test-rate-limit-wf-%d"
3508
wt := "sched-test-rate-limit-wt"
3509
3510
var runs int32
3511
workflowFn := func(ctx workflow.Context) error {
3512
workflow.SideEffect(ctx, func(ctx workflow.Context) any {
3513
atomic.AddInt32(&runs, 1)
3514
return 0
3515
})
3516
return nil
3517
}
3518
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
3519
3520
// create 10 copies of the schedule
3521
for i := range 10 {
3522
schedule := &schedulepb.Schedule{
3523
Spec: &schedulepb.ScheduleSpec{
3524
Interval: []*schedulepb.IntervalSpec{
3525
{Interval: durationpb.New(1 * time.Second)},
3526
},
3527
},
3528
Action: &schedulepb.ScheduleAction{
3529
Action: &schedulepb.ScheduleAction_StartWorkflow{
3530
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3531
WorkflowId: fmt.Sprintf(wid, i),
3532
WorkflowType: &commonpb.WorkflowType{Name: wt},
3533
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3534
},
3535
},
3536
},
3537
}
3538
_, err := s.FrontendClient().CreateSchedule(newContext(s.Context()), &workflowservice.CreateScheduleRequest{
3539
Namespace: s.Namespace().String(),
3540
ScheduleId: fmt.Sprintf(sid, i),
3541
Schedule: schedule,
3542
Identity: "test",
3543
RequestId: uuid.NewString(),
3544
})
3545
s.NoError(err)
3546
}
3547
3548
time.Sleep(5 * time.Second) //nolint:forbidigo
3549
3550
// With no rate limit, we'd see 10/second == 50 workflows run. With a limit of 1/sec, we
3551
// expect to see around 5.
3552
s.Less(atomic.LoadInt32(&runs), int32(10))
3553
}
3554
3555
// testNextTimeCache only applies to V1.
3556
func testNextTimeCache(t *testing.T, newContext contextFactory) {
3557
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
3558
3559
sid := "sched-test-next-time-cache"
3560
wid := "sched-test-next-time-cache-wf"
3561
wt := "sched-test-next-time-cache-wt"
3562
3563
schedule := &schedulepb.Schedule{
3564
Spec: &schedulepb.ScheduleSpec{
3565
Interval: []*schedulepb.IntervalSpec{
3566
{Interval: durationpb.New(1 * time.Second)},
3567
},
3568
},
3569
Action: &schedulepb.ScheduleAction{
3570
Action: &schedulepb.ScheduleAction_StartWorkflow{
3571
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3572
WorkflowId: wid,
3573
WorkflowType: &commonpb.WorkflowType{Name: wt},
3574
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3575
},
3576
},
3577
},
3578
}
3579
req := &workflowservice.CreateScheduleRequest{
3580
Namespace: s.Namespace().String(),
3581
ScheduleId: sid,
3582
Schedule: schedule,
3583
Identity: "test",
3584
RequestId: uuid.NewString(),
3585
}
3586
3587
var runs atomic.Int32
3588
workflowFn := func(ctx workflow.Context) error {
3589
workflow.SideEffect(ctx, func(ctx workflow.Context) any {
3590
runs.Add(1)
3591
return 0
3592
})
3593
return nil
3594
}
3595
s.SdkWorker().RegisterWorkflowWithOptions(workflowFn, workflow.RegisterOptions{Name: wt})
3596
3597
_, err := s.FrontendClient().CreateSchedule(newContext(s.Context()), req)
3598
s.NoError(err)
3599
3600
// wait for at least 13 runs
3601
const count = 13
3602
s.Eventually(func() bool { return runs.Load() >= count }, (count+10)*time.Second, 500*time.Millisecond)
3603
3604
// there should be only four side effects for 13 runs, and only two mentioning "Next"
3605
// (cache refills)
3606
events := s.GetHistory(s.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: scheduler.WorkflowIDPrefix + sid})
3607
var sideEffects, nextTimeSideEffects int
3608
for _, e := range events {
3609
if marker := e.GetMarkerRecordedEventAttributes(); marker.GetMarkerName() == "SideEffect" {
3610
sideEffects++
3611
if p, ok := marker.Details["data"]; ok && len(p.Payloads) == 1 {
3612
if string(p.Payloads[0].Metadata["messageType"]) == "temporal.server.api.schedule.v1.NextTimeCache" ||
3613
strings.Contains(payloads.ToString(p), `"Next"`) {
3614
nextTimeSideEffects++
3615
}
3616
}
3617
}
3618
}
3619
3620
const (
3621
// These match the ones in the scheduler workflow, but they're not exported.
3622
// Change these if those change.
3623
FutureActionCountForList = 5
3624
NextTimeCacheV2Size = 14
3625
3626
// Calculate expected results
3627
expectedCacheSize = NextTimeCacheV2Size - FutureActionCountForList + 1
3628
expectedRefills = (count + expectedCacheSize - 1) / expectedCacheSize
3629
uuidCacheRefills = (count + 9) / 10
3630
)
3631
s.Equal(expectedRefills+uuidCacheRefills, sideEffects)
3632
s.Equal(expectedRefills, nextTimeSideEffects)
3633
}
3634
3635
// getScheduleEntryFromVisibility polls visibility using ListSchedules until it finds a schedule
3636
// with the given id and for which the optional predicate function returns true.
3637
func getScheduleEntryFromVisibility(env testcore.Env, sid string, newContext contextFactory, predicate func(*schedulepb.ScheduleListEntry) bool) *schedulepb.ScheduleListEntry {
3638
env.T().Helper()
3639
var slEntry *schedulepb.ScheduleListEntry
3640
require.Eventually(env.T(), func() bool { // wait for visibility
3641
listResp, err := env.FrontendClient().ListSchedules(newContext(env.Context()), &workflowservice.ListSchedulesRequest{
3642
Namespace: env.Namespace().String(),
3643
MaximumPageSize: 5,
3644
})
3645
if err != nil {
3646
return false
3647
}
3648
for _, ent := range listResp.Schedules {
3649
if ent.ScheduleId == sid {
3650
if predicate != nil && !predicate(ent) {
3651
return false
3652
}
3653
slEntry = ent
3654
return true
3655
}
3656
}
3657
return false
3658
}, 15*time.Second, 1*time.Second)
3659
return slEntry
3660
}
3661
3662
func durationNear(t *testing.T, value, target time.Duration) {
3663
t.Helper()
3664
const tolerance = 5 * time.Second
3665
require.Greater(t, value, target-tolerance)
3666
require.Less(t, value, target+tolerance)
3667
}
3668
3669
func assertSameRecentActions(
3670
t *testing.T,
3671
expected *workflowservice.DescribeScheduleResponse, actual *schedulepb.ScheduleListEntry,
3672
) {
3673
t.Helper()
3674
if len(expected.Info.RecentActions) != len(actual.Info.RecentActions) {
3675
t.Fatalf(
3676
"RecentActions have different length expected %d, got %d",
3677
len(expected.Info.RecentActions),
3678
len(actual.Info.RecentActions))
3679
}
3680
for i := range expected.Info.RecentActions {
3681
if !proto.Equal(expected.Info.RecentActions[i], actual.Info.RecentActions[i]) {
3682
t.Errorf(
3683
"RecentActions are differ at index %d. Expected %v, got %v",
3684
i,
3685
expected.Info.RecentActions[i],
3686
actual.Info.RecentActions[i],
3687
)
3688
}
3689
}
3690
}
3691
3692
// assertRecentActionsNoDuplicateRunIDs verifies that no two entries in
3693
// RecentActions refer to the same workflow run. Duplicates can occur if the
3694
// migration between V1 and V2 schedulers doesn't properly deduplicate entries
3695
// that appear in both RunningWorkflows and RecentActions.
3696
func assertRecentActionsNoDuplicateRunIDs(t *testing.T, actions []*schedulepb.ScheduleActionResult) {
3697
t.Helper()
3698
seen := make(map[string]int) // runId -> index of first occurrence
3699
for i, action := range actions {
3700
runID := action.GetStartWorkflowResult().GetRunId()
3701
if runID == "" {
3702
continue
3703
}
3704
if firstIdx, ok := seen[runID]; ok {
3705
t.Errorf(
3706
"duplicate RunId %q in RecentActions at indices %d and %d (workflowId=%q)",
3707
runID, firstIdx, i, action.GetStartWorkflowResult().GetWorkflowId(),
3708
)
3709
}
3710
seen[runID] = i
3711
}
3712
}
3713
func testUpdateScheduleMemo(t *testing.T, newContext contextFactory) {
3714
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
3715
3716
sid := "sched-test-update-memo"
3717
wid := "sched-test-update-memo-wf"
3718
wt := "sched-test-update-memo-wt"
3719
3720
s.SdkWorker().RegisterWorkflowWithOptions(
3721
func(ctx workflow.Context) error { return nil },
3722
workflow.RegisterOptions{Name: wt},
3723
)
3724
3725
schedule := &schedulepb.Schedule{
3726
Spec: &schedulepb.ScheduleSpec{
3727
Interval: []*schedulepb.IntervalSpec{
3728
{Interval: durationpb.New(1 * time.Hour)},
3729
},
3730
},
3731
Action: &schedulepb.ScheduleAction{
3732
Action: &schedulepb.ScheduleAction_StartWorkflow{
3733
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3734
WorkflowId: wid,
3735
WorkflowType: &commonpb.WorkflowType{Name: wt},
3736
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3737
},
3738
},
3739
},
3740
}
3741
3742
memo1 := payload.EncodeString("val1")
3743
memo2 := payload.EncodeString("val2")
3744
3745
// Create schedule with initial memo.
3746
ctx := newContext(s.Context())
3747
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
3748
Namespace: s.Namespace().String(),
3749
ScheduleId: sid,
3750
Schedule: schedule,
3751
Identity: "test",
3752
RequestId: uuid.NewString(),
3753
Memo: &commonpb.Memo{
3754
Fields: map[string]*commonpb.Payload{
3755
"key1": memo1,
3756
"key2": memo2,
3757
},
3758
},
3759
})
3760
require.NoError(t, err)
3761
3762
// Verify initial memo.
3763
describeResp, err := s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
3764
Namespace: s.Namespace().String(),
3765
ScheduleId: sid,
3766
})
3767
require.NoError(t, err)
3768
require.Equal(t, memo1.Data, describeResp.Memo.Fields["key1"].Data)
3769
require.Equal(t, memo2.Data, describeResp.Memo.Fields["key2"].Data)
3770
3771
// Update: replace memo with only key3 (key1 and key2 should be gone).
3772
memo3 := payload.EncodeString("new")
3773
_, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{
3774
Namespace: s.Namespace().String(),
3775
ScheduleId: sid,
3776
Schedule: schedule,
3777
Identity: "test",
3778
RequestId: uuid.NewString(),
3779
Memo: &commonpb.Memo{
3780
Fields: map[string]*commonpb.Payload{
3781
"key3": memo3,
3782
},
3783
},
3784
})
3785
require.NoError(t, err)
3786
3787
// Verify replaced memo.
3788
describeResp, err = s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
3789
Namespace: s.Namespace().String(),
3790
ScheduleId: sid,
3791
})
3792
require.NoError(t, err)
3793
require.Nil(t, describeResp.Memo.Fields["key1"], "key1 should be gone after replace")
3794
require.Nil(t, describeResp.Memo.Fields["key2"], "key2 should be gone after replace")
3795
require.Equal(t, memo3.Data, describeResp.Memo.Fields["key3"].Data, "key3 should be set")
3796
3797
// Update with nil memo (no change).
3798
_, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{
3799
Namespace: s.Namespace().String(),
3800
ScheduleId: sid,
3801
Schedule: schedule,
3802
Identity: "test",
3803
RequestId: uuid.NewString(),
3804
})
3805
require.NoError(t, err)
3806
3807
// Verify memo unchanged.
3808
describeResp, err = s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
3809
Namespace: s.Namespace().String(),
3810
ScheduleId: sid,
3811
})
3812
require.NoError(t, err)
3813
require.Equal(t, memo3.Data, describeResp.Memo.Fields["key3"].Data, "key3 should be unchanged")
3814
3815
// Update with empty memo (clear all).
3816
_, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{
3817
Namespace: s.Namespace().String(),
3818
ScheduleId: sid,
3819
Schedule: schedule,
3820
Identity: "test",
3821
RequestId: uuid.NewString(),
3822
Memo: &commonpb.Memo{
3823
Fields: map[string]*commonpb.Payload{},
3824
},
3825
})
3826
require.NoError(t, err)
3827
3828
// Verify memo cleared.
3829
describeResp, err = s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
3830
Namespace: s.Namespace().String(),
3831
ScheduleId: sid,
3832
})
3833
require.NoError(t, err)
3834
require.Empty(t, describeResp.Memo.GetFields(), "memo should be empty after replace with empty map")
3835
}
3836
3837
func testUpdateScheduleMemoRejected(t *testing.T, newContext contextFactory) {
3838
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
3839
3840
sid := "sched-test-update-memo-rejected"
3841
wid := "sched-test-update-memo-rejected-wf"
3842
wt := "sched-test-update-memo-rejected-wt"
3843
3844
s.SdkWorker().RegisterWorkflowWithOptions(
3845
func(ctx workflow.Context) error { return nil },
3846
workflow.RegisterOptions{Name: wt},
3847
)
3848
3849
schedule := &schedulepb.Schedule{
3850
Spec: &schedulepb.ScheduleSpec{
3851
Interval: []*schedulepb.IntervalSpec{
3852
{Interval: durationpb.New(1 * time.Hour)},
3853
},
3854
},
3855
Action: &schedulepb.ScheduleAction{
3856
Action: &schedulepb.ScheduleAction_StartWorkflow{
3857
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3858
WorkflowId: wid,
3859
WorkflowType: &commonpb.WorkflowType{Name: wt},
3860
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3861
},
3862
},
3863
},
3864
}
3865
3866
// Create V1 schedule.
3867
ctx := newContext(s.Context())
3868
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
3869
Namespace: s.Namespace().String(),
3870
ScheduleId: sid,
3871
Schedule: schedule,
3872
Identity: "test",
3873
RequestId: uuid.NewString(),
3874
})
3875
require.NoError(t, err)
3876
3877
// Update with memo should be rejected.
3878
_, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{
3879
Namespace: s.Namespace().String(),
3880
ScheduleId: sid,
3881
Schedule: schedule,
3882
Identity: "test",
3883
RequestId: uuid.NewString(),
3884
Memo: &commonpb.Memo{
3885
Fields: map[string]*commonpb.Payload{
3886
"key": payload.EncodeString("value"),
3887
},
3888
},
3889
})
3890
require.Error(t, err)
3891
var failedPrecondition *serviceerror.FailedPrecondition
3892
require.ErrorAs(t, err, &failedPrecondition)
3893
require.Contains(t, err.Error(), "memo updates are not supported on workflow-backed schedules")
3894
}
3895
3896
func testUpdateScheduleMemoOnly(t *testing.T, newContext contextFactory) {
3897
// UpdateScheduleRequest uses replace semantics for the schedule field, so omitting it
3898
// causes the schedule to be unset. Memo-only updates require the server to skip replacing
3899
// the schedule when the field is nil, similar to how memo and search_attributes are handled.
3900
t.Skip("memo-only updates not yet supported: omitting the schedule field unsets the schedule")
3901
3902
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
3903
3904
sid := "sched-test-update-memo-only"
3905
wid := "sched-test-update-memo-only-wf"
3906
wt := "sched-test-update-memo-only-wt"
3907
3908
s.SdkWorker().RegisterWorkflowWithOptions(
3909
func(ctx workflow.Context) error { return nil },
3910
workflow.RegisterOptions{Name: wt},
3911
)
3912
3913
schedule := &schedulepb.Schedule{
3914
Spec: &schedulepb.ScheduleSpec{
3915
Interval: []*schedulepb.IntervalSpec{
3916
{Interval: durationpb.New(1 * time.Hour)},
3917
},
3918
},
3919
Action: &schedulepb.ScheduleAction{
3920
Action: &schedulepb.ScheduleAction_StartWorkflow{
3921
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3922
WorkflowId: wid,
3923
WorkflowType: &commonpb.WorkflowType{Name: wt},
3924
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
3925
},
3926
},
3927
},
3928
}
3929
3930
// Create schedule with initial memo.
3931
memo1 := payload.EncodeString("val1")
3932
ctx := newContext(s.Context())
3933
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
3934
Namespace: s.Namespace().String(),
3935
ScheduleId: sid,
3936
Schedule: schedule,
3937
Identity: "test",
3938
RequestId: uuid.NewString(),
3939
Memo: &commonpb.Memo{
3940
Fields: map[string]*commonpb.Payload{"key1": memo1},
3941
},
3942
})
3943
require.NoError(t, err)
3944
3945
// Update only memo, without setting the schedule field.
3946
memo2 := payload.EncodeString("val2")
3947
_, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{
3948
Namespace: s.Namespace().String(),
3949
ScheduleId: sid,
3950
Identity: "test",
3951
RequestId: uuid.NewString(),
3952
Memo: &commonpb.Memo{
3953
Fields: map[string]*commonpb.Payload{"key1": memo2},
3954
},
3955
})
3956
require.NoError(t, err)
3957
3958
// Verify memo was updated and schedule is still intact.
3959
describeResp, err := s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{
3960
Namespace: s.Namespace().String(),
3961
ScheduleId: sid,
3962
})
3963
require.NoError(t, err)
3964
require.Equal(t, memo2.Data, describeResp.Memo.Fields["key1"].Data, "memo should be updated")
3965
require.NotNil(t, describeResp.Schedule.Spec, "schedule spec should not be nil")
3966
require.NotEmpty(t, describeResp.Schedule.Spec.Interval, "schedule spec intervals should be preserved")
3967
require.NotNil(t, describeResp.Schedule.Action, "schedule action should be preserved")
3968
}
3969
3970
func testCHASMUnpauseResumesProcessing(t *testing.T, newContext contextFactory) {
3971
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
3972
3973
sid := "sched-test-unpause-resumes"
3974
wid := "sched-test-unpause-resumes-wf"
3975
wt := "sched-test-unpause-resumes-wt"
3976
3977
var runs int32
3978
s.SdkWorker().RegisterWorkflowWithOptions(
3979
func(ctx workflow.Context) error {
3980
workflow.SideEffect(ctx, func(ctx workflow.Context) any {
3981
atomic.AddInt32(&runs, 1)
3982
return 0
3983
})
3984
return nil
3985
},
3986
workflow.RegisterOptions{Name: wt},
3987
)
3988
3989
_, err := s.FrontendClient().CreateSchedule(newContext(s.Context()), &workflowservice.CreateScheduleRequest{
3990
Namespace: s.Namespace().String(),
3991
ScheduleId: sid,
3992
Schedule: &schedulepb.Schedule{
3993
Spec: &schedulepb.ScheduleSpec{
3994
Interval: []*schedulepb.IntervalSpec{{Interval: durationpb.New(1 * time.Second)}},
3995
},
3996
Action: &schedulepb.ScheduleAction{
3997
Action: &schedulepb.ScheduleAction_StartWorkflow{
3998
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
3999
WorkflowId: wid,
4000
WorkflowType: &commonpb.WorkflowType{Name: wt},
4001
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4002
},
4003
},
4004
},
4005
},
4006
Identity: "test",
4007
RequestId: uuid.NewString(),
4008
})
4009
s.NoError(err)
4010
4011
// Wait for the schedule to fire at least once, confirming it's running.
4012
s.Eventually(func() bool { return atomic.LoadInt32(&runs) >= 1 }, 15*time.Second, 500*time.Millisecond)
4013
4014
// Pause.
4015
_, err = s.FrontendClient().PatchSchedule(newContext(s.Context()), &workflowservice.PatchScheduleRequest{
4016
Namespace: s.Namespace().String(),
4017
ScheduleId: sid,
4018
Patch: &schedulepb.SchedulePatch{Pause: "pausing for test"},
4019
Identity: "test",
4020
RequestId: uuid.NewString(),
4021
})
4022
s.NoError(err)
4023
4024
// Wait for the already-queued generator task to run after pause. That task
4025
// observes paused state, performs no-op scheduling, and then the schedule
4026
// becomes quiescent (no new runs over a stability window).
4027
stableSamples := 0
4028
lastRuns := atomic.LoadInt32(&runs)
4029
s.Eventually(func() bool {
4030
currentRuns := atomic.LoadInt32(&runs)
4031
if currentRuns == lastRuns {
4032
stableSamples++
4033
} else {
4034
lastRuns = currentRuns
4035
stableSamples = 0
4036
}
4037
return stableSamples >= 6
4038
}, 15*time.Second, 500*time.Millisecond)
4039
runsBeforeUnpause := atomic.LoadInt32(&runs)
4040
4041
// Unpause.
4042
_, err = s.FrontendClient().PatchSchedule(newContext(s.Context()), &workflowservice.PatchScheduleRequest{
4043
Namespace: s.Namespace().String(),
4044
ScheduleId: sid,
4045
Patch: &schedulepb.SchedulePatch{Unpause: "resuming"},
4046
Identity: "test",
4047
RequestId: uuid.NewString(),
4048
})
4049
s.NoError(err)
4050
4051
// The generator should be kicked immediately on unpause and new runs should follow.
4052
s.Eventually(
4053
func() bool { return atomic.LoadInt32(&runs) > runsBeforeUnpause },
4054
15*time.Second,
4055
500*time.Millisecond,
4056
"schedule should resume processing after unpause",
4057
)
4058
}
4059
4060
// testPausedDropsCatchup verifies that an action scheduled by the spec during
4061
// a paused window is NOT invoked when the schedule is unpaused.
4062
func testPausedDropsCatchup(t *testing.T, newContext contextFactory) {
4063
s := newEnvWithIdleTime(t, shortIdleTime)
4064
4065
sid := testcore.RandomizeStr("sched-paused-drops-catchup")
4066
wid := testcore.RandomizeStr("sched-paused-drops-catchup-wf")
4067
wt := testcore.RandomizeStr("sched-paused-drops-catchup-wt")
4068
4069
var runs atomic.Int32
4070
registerCountingWorkflow(s, wt, &runs)
4071
4072
// Single calendar entry a few seconds in the future. While paused, its
4073
// time will pass. Offset must exceed worst-case CreateSchedule latency
4074
// under parallel load so the entry is still genuinely in the future when
4075
// the server materializes the schedule - otherwise the test passes for
4076
// the wrong reason (HWM already past the entry at create time).
4077
fireAt := time.Now().Add(10 * time.Second).UTC()
4078
ctx := newContext(s.Context())
4079
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4080
Spec: &schedulepb.ScheduleSpec{
4081
Calendar: []*schedulepb.CalendarSpec{calendarSpec(fireAt)},
4082
},
4083
State: &schedulepb.ScheduleState{Paused: true},
4084
Action: startWorkflowAction(s, wid, wt),
4085
})
4086
4087
await.RequireTruef(t, func() bool {
4088
desc, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4089
Namespace: s.Namespace().String(),
4090
ScheduleId: sid,
4091
})
4092
return descErr == nil && len(desc.Info.FutureActionTimes) == 0
4093
}, awaitTimeout, pollInterval,
4094
"FutureActionTimes should empty out once the only calendar date passes (proves HWM advanced past it while paused)")
4095
4096
patchSchedule(ctx, t, s, sid, &schedulepb.SchedulePatch{Unpause: "drops-catchup-test"})
4097
await.RequireTruef(t, func() bool { return scheduleClosed(ctx, s, sid) },
4098
awaitTimeout, pollInterval,
4099
"schedule should close from idle after unpause (no future actions, no replay)")
4100
}
4101
4102
// testPausedScheduleNeverIdles verifies that a paused schedule is held open
4103
// indefinitely on both backends, even past the configured idle window.
4104
func testPausedScheduleNeverIdles(t *testing.T, newContext contextFactory) {
4105
s := newEnvWithIdleTime(t, shortIdleTime)
4106
4107
sid := testcore.RandomizeStr("sched-paused-never-idles")
4108
wid := testcore.RandomizeStr("sched-paused-never-idles-wf")
4109
wt := testcore.RandomizeStr("sched-paused-never-idles-wt")
4110
4111
var runs atomic.Int32
4112
registerCountingWorkflow(s, wt, &runs)
4113
4114
ctx := newContext(s.Context())
4115
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4116
Spec: intervalSpec(fastInterval),
4117
Action: startWorkflowAction(s, wid, wt),
4118
})
4119
4120
await.RequireTruef(t,
4121
func() bool { return runs.Load() >= 1 },
4122
awaitTimeout, pollInterval,
4123
"schedule should have fired at least once before pause",
4124
)
4125
4126
patchSchedule(ctx, t, s, sid, &schedulepb.SchedulePatch{Pause: "never-idles-test"})
4127
4128
// Across a window well past IdleTime, the schedule must never idle-close.
4129
require.Never(t, func() bool { return scheduleClosed(ctx, s, sid) },
4130
3*shortIdleTime, pollInterval,
4131
"paused schedule must not close from idle even past IdleTime")
4132
4133
desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4134
Namespace: s.Namespace().String(),
4135
ScheduleId: sid,
4136
})
4137
require.NoError(t, err)
4138
require.True(t, desc.Schedule.State.Paused, "paused schedule must stay paused")
4139
4140
// Also verify by unpausing and seeing actions resume.
4141
runsBeforeUnpause := runs.Load()
4142
patchSchedule(ctx, t, s, sid, &schedulepb.SchedulePatch{Unpause: "never-idles-test-resume"})
4143
await.RequireTruef(t,
4144
func() bool { return runs.Load() > runsBeforeUnpause },
4145
awaitTimeout, pollInterval,
4146
"paused-then-unpaused schedule should resume firing (it should not have closed)",
4147
)
4148
}
4149
4150
// testPausedEmptySpecStaysOpen verifies that a schedule created with an empty
4151
// spec (no Calendar / CronString / Interval) and Paused=true - the SDK
4152
// manual-only pattern - can be created without timing out and remains
4153
// describable.
4154
func testPausedEmptySpecStaysOpen(t *testing.T, newContext contextFactory) {
4155
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4156
4157
sid := testcore.RandomizeStr("sched-paused-empty-spec")
4158
wid := testcore.RandomizeStr("sched-paused-empty-spec-wf")
4159
wt := testcore.RandomizeStr("sched-paused-empty-spec-wt")
4160
4161
var runs atomic.Int32
4162
registerCountingWorkflow(s, wt, &runs)
4163
4164
// Empty spec + paused: a manual-only schedule. Create must succeed without
4165
// timing out (the original regression was "context deadline exceeded").
4166
ctx := newContext(s.Context())
4167
createCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
4168
defer cancel()
4169
createSchedule(createCtx, t, s, sid, &schedulepb.Schedule{
4170
Spec: &schedulepb.ScheduleSpec{},
4171
State: &schedulepb.ScheduleState{Paused: true},
4172
Action: startWorkflowAction(s, wid, wt),
4173
})
4174
4175
require.Never(t, func() bool { return runs.Load() > 0 },
4176
neverWindow, pollInterval,
4177
"empty paused spec must not fire any actions automatically")
4178
4179
desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4180
Namespace: s.Namespace().String(),
4181
ScheduleId: sid,
4182
})
4183
require.NoError(t, err)
4184
require.True(t, desc.Schedule.State.Paused, "schedule must still be paused")
4185
4186
// Unpause + TriggerImmediately to sanity-check the schedule is functional,
4187
// not just open.
4188
patchSchedule(ctx, t, s, sid, &schedulepb.SchedulePatch{
4189
Unpause: "empty-spec-trigger",
4190
TriggerImmediately: &schedulepb.TriggerImmediatelyRequest{},
4191
})
4192
4193
await.RequireTruef(t,
4194
func() bool { return runs.Load() == 1 },
4195
awaitTimeout, pollInterval,
4196
"manual trigger after unpause should fire exactly one action",
4197
)
4198
}
4199
4200
// testTriggerImmediatelyOnActiveSchedule verifies that a TriggerImmediately
4201
// patch on a running schedule fires an extra action and leaves the schedule
4202
// active.
4203
func testTriggerImmediatelyOnActiveSchedule(t *testing.T, newContext contextFactory) {
4204
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4205
4206
sid := testcore.RandomizeStr("sched-trigger-on-active")
4207
wid := testcore.RandomizeStr("sched-trigger-on-active-wf")
4208
wt := testcore.RandomizeStr("sched-trigger-on-active-wt")
4209
4210
var runs atomic.Int32
4211
registerCountingWorkflow(s, wt, &runs)
4212
4213
ctx := newContext(s.Context())
4214
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4215
// Fire a year into the future, keeping the schedule active, but without firing
4216
// actions.
4217
Spec: &schedulepb.ScheduleSpec{
4218
Calendar: []*schedulepb.CalendarSpec{calendarSpec(time.Now().AddDate(1, 0, 0).UTC())},
4219
},
4220
Action: startWorkflowAction(s, wid, wt),
4221
Policies: &schedulepb.SchedulePolicies{
4222
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
4223
},
4224
})
4225
4226
await.RequireTruef(t, func() bool {
4227
desc, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4228
Namespace: s.Namespace().String(),
4229
ScheduleId: sid,
4230
})
4231
return descErr == nil && len(desc.Info.FutureActionTimes) > 0
4232
}, awaitTimeout, pollInterval, "schedule should reach active state with future actions planned")
4233
require.Zero(t, runs.Load(), "no automated action should fire before the trigger")
4234
4235
patchSchedule(ctx, t, s, sid, triggerPatch(enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL))
4236
await.RequireTruef(t,
4237
func() bool { return runs.Load() == 1 },
4238
awaitTimeout, pollInterval,
4239
"TriggerImmediately should fire exactly one action on the active schedule",
4240
)
4241
4242
desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4243
Namespace: s.Namespace().String(),
4244
ScheduleId: sid,
4245
})
4246
require.NoError(t, err)
4247
require.NotEmpty(t, desc.Info.RecentActions, "RecentActions should include the manual trigger")
4248
require.NotEmpty(t, desc.Info.FutureActionTimes, "schedule should still have future automated actions planned")
4249
}
4250
4251
// testTriggerImmediatelyOnPausedSchedule verifies that TriggerImmediately fires
4252
// an action even when the schedule is paused. Manual starts bypass the paused
4253
// gate via useScheduledAction's Manual carve-out in processBuffer.
4254
func testTriggerImmediatelyOnPausedSchedule(t *testing.T, newContext contextFactory) {
4255
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4256
4257
sid := testcore.RandomizeStr("sched-trigger-on-paused")
4258
wid := testcore.RandomizeStr("sched-trigger-on-paused-wf")
4259
wt := testcore.RandomizeStr("sched-trigger-on-paused-wt")
4260
4261
var runs atomic.Int32
4262
registerCountingWorkflow(s, wt, &runs)
4263
4264
ctx := newContext(s.Context())
4265
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4266
Spec: intervalSpec(fastInterval),
4267
State: &schedulepb.ScheduleState{Paused: true},
4268
Action: startWorkflowAction(s, wid, wt),
4269
})
4270
4271
// Paused suppresses firing even though the 1s interval would otherwise tick.
4272
require.Never(t, func() bool { return runs.Load() > 0 },
4273
neverWindow, pollInterval,
4274
"paused schedule must not fire automated actions before the trigger")
4275
4276
patchSchedule(ctx, t, s, sid, triggerPatch(enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL))
4277
4278
await.RequireTruef(t, func() bool { return runs.Load() == 1 },
4279
awaitTimeout, pollInterval,
4280
"TriggerImmediately must fire exactly one action despite the schedule being paused")
4281
4282
desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4283
Namespace: s.Namespace().String(),
4284
ScheduleId: sid,
4285
})
4286
require.NoError(t, err)
4287
require.NotEmpty(t, desc.Info.RecentActions, "RecentActions should include the manual trigger")
4288
require.True(t, desc.Schedule.State.Paused, "schedule must still be paused after trigger fires")
4289
}
4290
4291
// testTriggerImmediatelyAfterActionsExhausted verifies that TriggerImmediately
4292
// fires an action even on a schedule that has no LimitedActions slots left.
4293
func testTriggerImmediatelyAfterActionsExhausted(t *testing.T, newContext contextFactory) {
4294
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4295
4296
sid := testcore.RandomizeStr("sched-trigger-after-exhausted")
4297
wid := testcore.RandomizeStr("sched-trigger-after-exhausted-wf")
4298
wt := testcore.RandomizeStr("sched-trigger-after-exhausted-wt")
4299
4300
var runs atomic.Int32
4301
registerCountingWorkflow(s, wt, &runs)
4302
4303
ctx := newContext(s.Context())
4304
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4305
Spec: intervalSpec(fastInterval),
4306
// Start exhausted so no automated action fires.
4307
State: &schedulepb.ScheduleState{LimitedActions: true, RemainingActions: 0},
4308
Action: startWorkflowAction(s, wid, wt),
4309
})
4310
4311
require.Never(t, func() bool { return runs.Load() > 0 },
4312
neverWindow, pollInterval,
4313
"exhausted schedule must not auto-fire before the trigger")
4314
4315
patchSchedule(ctx, t, s, sid, triggerPatch(enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL))
4316
await.RequireTruef(t, func() bool { return runs.Load() == 1 },
4317
awaitTimeout, pollInterval,
4318
"TriggerImmediately must fire exactly once despite RemainingActions=0")
4319
4320
desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4321
Namespace: s.Namespace().String(),
4322
ScheduleId: sid,
4323
})
4324
require.NoError(t, err)
4325
require.NotEmpty(t, desc.Info.RecentActions, "RecentActions should include the manual trigger")
4326
require.Equal(t, int64(0), desc.Schedule.State.RemainingActions,
4327
"manual trigger must not consume a LimitedActions slot")
4328
}
4329
4330
func testBackfillReprocessesCompletedAction(
4331
t *testing.T,
4332
newContext contextFactory,
4333
paused bool,
4334
intervalsOnEachSide int,
4335
) {
4336
s := testcore.NewEnv(t, scheduleCommonOpts(t)...)
4337
4338
sid := testcore.RandomizeStr("sched-backfill-reprocess")
4339
wid := testcore.RandomizeStr("sched-backfill-reprocess-wf")
4340
wt := testcore.RandomizeStr("sched-backfill-reprocess-wt")
4341
4342
var runs atomic.Int32
4343
registerCountingWorkflow(s, wt, &runs)
4344
4345
ctx := newContext(s.Context())
4346
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4347
Spec: intervalSpec(fastInterval),
4348
State: &schedulepb.ScheduleState{
4349
LimitedActions: true,
4350
RemainingActions: 1,
4351
},
4352
Action: startWorkflowAction(s, wid, wt),
4353
})
4354
4355
var completedTime time.Time
4356
await.RequireTruef(t, func() bool {
4357
desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4358
Namespace: s.Namespace().String(),
4359
ScheduleId: sid,
4360
})
4361
if err != nil {
4362
return false
4363
}
4364
for _, action := range desc.GetInfo().GetRecentActions() {
4365
if action.GetStartWorkflowStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED {
4366
completedTime = action.GetScheduleTime().AsTime()
4367
return true
4368
}
4369
}
4370
return false
4371
}, awaitTimeout, pollInterval, "the automatic action should complete")
4372
require.Equal(t, int32(1), runs.Load())
4373
4374
if paused {
4375
patchSchedule(ctx, t, s, sid, &schedulepb.SchedulePatch{Pause: "test completed-action backfill"})
4376
}
4377
4378
rangeOffset := time.Duration(intervalsOnEachSide) * fastInterval
4379
patchSchedule(ctx, t, s, sid, backfillPatch(
4380
completedTime.Add(-rangeOffset),
4381
completedTime.Add(rangeOffset),
4382
enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL,
4383
))
4384
4385
expectedRuns := int32(2 + 2*intervalsOnEachSide)
4386
await.RequireTruef(t, func() bool { return runs.Load() == expectedRuns }, neverWindow, pollInterval,
4387
"backfill should reprocess the completed action exactly once")
4388
4389
desc, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4390
Namespace: s.Namespace().String(),
4391
ScheduleId: sid,
4392
})
4393
require.NoError(t, err)
4394
require.Equal(t, int64(expectedRuns), desc.GetInfo().GetActionCount())
4395
require.Empty(t, desc.GetInfo().GetRunningWorkflows())
4396
require.Equal(t, paused, desc.GetSchedule().GetState().GetPaused())
4397
}
4398
4399
// testBackfillWithBufferOneOverlap pins the expected behavior of BUFFER_ONE
4400
// over a multi-tick backfill: the first start runs immediately, exactly one
4401
// follow-up is buffered (Attempt=-1 deferred), the rest are dropped, and the
4402
// deferred one runs once the first completes. Currently SKIPPED: fails on
4403
// both V1 and CHASM because the deferred start never gets re-enabled after
4404
// the running workflow completes. The first start fires, the rest never run.
4405
// Likely a real bug in the BUFFER_ONE + backfill (Manual=true) interaction -
4406
// recordCompletedAction's re-enable loop on Attempt==-1 may not be running
4407
// against backfill-buffered starts. Worth a separate investigation.
4408
func testBackfillWithBufferOneOverlap(t *testing.T, newContext contextFactory) {
4409
// TODO(temporalio/temporal): track removing this skip once the BUFFER_ONE
4410
// backfill deferred re-enable path is fixed. Verify by running:
4411
// go test ./tests/ -run 'TestScheduleCHASM/Backfill/BufferOneOverlap' -v
4412
t.Skip("BUFFER_ONE backfill deferred re-enable is broken on both V1 and CHASM; see test doc")
4413
4414
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4415
4416
sid := testcore.RandomizeStr("sched-backfill-buffer-one")
4417
wid := testcore.RandomizeStr("sched-backfill-buffer-one-wf")
4418
wt := testcore.RandomizeStr("sched-backfill-buffer-one-wt")
4419
4420
// Gate runs so the first-running / one-deferred / rest-dropped sequence is deterministic.
4421
var runs atomic.Int32
4422
registerGatedWorkflow(s, wt, &runs)
4423
4424
ctx := newContext(s.Context())
4425
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4426
Spec: intervalSpec(fastInterval),
4427
State: &schedulepb.ScheduleState{Paused: true},
4428
Action: startWorkflowAction(s, wid, wt),
4429
})
4430
4431
now := time.Now().UTC()
4432
patchSchedule(ctx, t, s, sid, backfillPatch(now.Add(-5*time.Second), now, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE))
4433
4434
// First backfill start runs with exactly one buffered behind it (BUFFER_ONE drops the rest).
4435
await.RequireTruef(t, func() bool {
4436
desc, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4437
Namespace: s.Namespace().String(),
4438
ScheduleId: sid,
4439
})
4440
return descErr == nil && desc.GetInfo().GetBufferSize() == 1 && len(desc.GetInfo().GetRunningWorkflows()) == 1
4441
}, awaitTimeout, pollInterval, "expected exactly one running backfill start with one deferred behind it")
4442
require.Equal(t, int32(1), runs.Load(), "only the first backfill start should have fired so far")
4443
4444
// Releasing the running start must re-enable the deferred one (Attempt=-1 -> 0) so it fires.
4445
require.Equal(t, 1, completeRunningWorkflows(ctx, t, s, sid))
4446
await.RequireTruef(t, func() bool { return runs.Load() == 2 },
4447
awaitTimeout, pollInterval,
4448
"deferred backfill start must fire after the running one completes (Attempt=-1 -> 0 re-enable)")
4449
require.Never(t, func() bool { return runs.Load() > 2 },
4450
neverWindow, pollInterval,
4451
"BUFFER_ONE must collapse the rest of the backfill ticks - only the deferred start re-enables")
4452
}
4453
4454
// testBackfillRangeSmallerThanInterval covers the edge where a backfill range
4455
// is narrower than the spec interval - no spec tick lands inside it, so no
4456
// actions fire and the backfiller still drains cleanly.
4457
func testBackfillRangeSmallerThanInterval(t *testing.T, newContext contextFactory) {
4458
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4459
4460
sid := testcore.RandomizeStr("sched-backfill-narrow")
4461
wid := testcore.RandomizeStr("sched-backfill-narrow-wf")
4462
wt := testcore.RandomizeStr("sched-backfill-narrow-wt")
4463
4464
var runs atomic.Int32
4465
registerCountingWorkflow(s, wt, &runs)
4466
4467
ctx := newContext(s.Context())
4468
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4469
// noOpInterval (1h) ticks on the hour; the backfill window below is mid-hour.
4470
Spec: intervalSpec(noOpInterval),
4471
State: &schedulepb.ScheduleState{Paused: true},
4472
Action: startWorkflowAction(s, wid, wt),
4473
})
4474
4475
// A 10s window (above the 1s resolution, below the 1h interval) anchored mid-hour
4476
// in the previous hour: always in the past and provably free of an on-the-hour tick.
4477
prevHour := time.Now().UTC().Truncate(time.Hour).Add(-time.Hour)
4478
windowStart := prevHour.Add(20 * time.Minute)
4479
windowEnd := windowStart.Add(10 * time.Second)
4480
patchSchedule(ctx, t, s, sid, backfillPatch(windowStart, windowEnd, enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL))
4481
4482
// No spec tick falls inside a sub-interval window, so no action fires.
4483
require.Never(t, func() bool { return runs.Load() > 0 },
4484
neverWindow, pollInterval,
4485
"backfill range narrower than spec interval must produce no actions")
4486
4487
// And the schedule must still be describable - the backfiller drained without
4488
// firing anything and got cleaned up.
4489
_, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4490
Namespace: s.Namespace().String(),
4491
ScheduleId: sid,
4492
})
4493
require.NoError(t, err, "schedule must remain describable after empty backfill")
4494
}
4495
4496
// testBackfillWithSkipOverlap verifies that SKIP overlap correctly collapses a
4497
// multi-tick backfill range to a single workflow execution.
4498
func testBackfillWithSkipOverlap(t *testing.T, newContext contextFactory) {
4499
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4500
4501
sid := testcore.RandomizeStr("sched-backfill-skip")
4502
wid := testcore.RandomizeStr("sched-backfill-skip-wf")
4503
wt := testcore.RandomizeStr("sched-backfill-skip-wt")
4504
4505
var runs atomic.Int32
4506
registerCountingWorkflow(s, wt, &runs)
4507
4508
ctx := newContext(s.Context())
4509
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4510
Spec: intervalSpec(fastInterval),
4511
State: &schedulepb.ScheduleState{Paused: true},
4512
Action: startWorkflowAction(s, wid, wt),
4513
})
4514
4515
now := time.Now().UTC()
4516
patchSchedule(ctx, t, s, sid, backfillPatch(now.Add(-5*time.Second), now, enumspb.SCHEDULE_OVERLAP_POLICY_SKIP))
4517
4518
// SKIP collapses the 5-tick backfill to exactly one fire, and it stays there.
4519
await.RequireTruef(t, func() bool { return runs.Load() == 1 },
4520
awaitTimeout, pollInterval,
4521
"SKIP backfill should fire exactly once")
4522
require.Never(t, func() bool { return runs.Load() > 1 },
4523
neverWindow, pollInterval,
4524
"SKIP backfill must collapse to a single execution, not fire all 5 ticks")
4525
}
4526
4527
func testUpdateScheduleRequestIDTooLong(t *testing.T, newContext contextFactory) {
4528
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4529
4530
sid := "sched-test-update-reqid-too-long"
4531
wid := "sched-test-update-reqid-too-long-wf"
4532
wt := "sched-test-update-reqid-too-long-wt"
4533
4534
s.SdkWorker().RegisterWorkflowWithOptions(
4535
func(ctx workflow.Context) error { return nil },
4536
workflow.RegisterOptions{Name: wt},
4537
)
4538
4539
schedule := &schedulepb.Schedule{
4540
Spec: intervalSpec(noOpInterval),
4541
Action: startWorkflowAction(s, wid, wt),
4542
}
4543
4544
ctx := newContext(s.Context())
4545
createSchedule(ctx, t, s, sid, schedule)
4546
4547
// Update with an oversized request ID.
4548
_, err := s.FrontendClient().UpdateSchedule(ctx, &workflowservice.UpdateScheduleRequest{
4549
Namespace: s.Namespace().String(),
4550
ScheduleId: sid,
4551
Schedule: schedule,
4552
Identity: "test",
4553
RequestId: strings.Repeat("x", 1001),
4554
})
4555
var invalidArgReqID *serviceerror.InvalidArgument
4556
require.ErrorAs(t, err, &invalidArgReqID)
4557
}
4558
4559
func testUpdateScheduleBlobSizeLimit(t *testing.T, newContext contextFactory) {
4560
s := newScheduleEnv(t,
4561
append(scheduleCommonOpts(t),
4562
testcore.WithDynamicConfig(dynamicconfig.BlobSizeLimitError, 1000),
4563
testcore.WithDynamicConfig(dynamicconfig.BlobSizeLimitWarn, 500),
4564
)...,
4565
)
4566
4567
sid := "sched-test-update-blob-limit"
4568
wid := "sched-test-update-blob-limit-wf"
4569
wt := "sched-test-update-blob-limit-wt"
4570
4571
s.SdkWorker().RegisterWorkflowWithOptions(
4572
func(ctx workflow.Context) error { return nil },
4573
workflow.RegisterOptions{Name: wt},
4574
)
4575
4576
schedule := &schedulepb.Schedule{
4577
Spec: &schedulepb.ScheduleSpec{
4578
Interval: []*schedulepb.IntervalSpec{
4579
{Interval: durationpb.New(1 * time.Hour)},
4580
},
4581
},
4582
Action: &schedulepb.ScheduleAction{
4583
Action: &schedulepb.ScheduleAction_StartWorkflow{
4584
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
4585
WorkflowId: wid,
4586
WorkflowType: &commonpb.WorkflowType{Name: wt},
4587
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4588
},
4589
},
4590
},
4591
}
4592
4593
// Create schedule.
4594
ctx := newContext(s.Context())
4595
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
4596
Namespace: s.Namespace().String(),
4597
ScheduleId: sid,
4598
Schedule: schedule,
4599
Identity: "test",
4600
RequestId: uuid.NewString(),
4601
})
4602
require.NoError(t, err)
4603
4604
// Update with an oversized memo that exceeds the blob size limit.
4605
largeMemo := &commonpb.Memo{
4606
Fields: map[string]*commonpb.Payload{
4607
"key": {Data: make([]byte, 1001)},
4608
},
4609
}
4610
_, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{
4611
Namespace: s.Namespace().String(),
4612
ScheduleId: sid,
4613
Schedule: schedule,
4614
Identity: "test",
4615
RequestId: uuid.NewString(),
4616
Memo: largeMemo,
4617
})
4618
var invalidArgBlob *serviceerror.InvalidArgument
4619
require.ErrorAs(t, err, &invalidArgBlob)
4620
}
4621
4622
// TestScheduleCreationRolloutPercent verifies that
4623
// CHASMSchedulerCreationRolloutPercent acts as a per-schedule sampling gate
4624
// after EnableCHASMSchedulerCreation is on: at 50%, two schedules whose IDs
4625
// bucket on opposite sides of the rollout land on different stacks.
4626
func TestScheduleCreationRolloutPercent(t *testing.T) {
4627
opts := append(scheduleCommonOpts(t),
4628
// V1 worker is needed because at 50% rollout some schedules land on V1.
4629
testcore.WithWorkerService("V1 scheduler"),
4630
testcore.WithDynamicConfig(dynamicconfig.EnableCHASMSchedulerCreation, true),
4631
testcore.WithDynamicConfig(dynamicconfig.CHASMSchedulerCreationRolloutPercent, 50),
4632
)
4633
s := newScheduleEnv(t, opts...)
4634
ctx := s.Context()
4635
nsName := s.Namespace().String()
4636
nsID := s.NamespaceID().String()
4637
4638
chasmSID, v1SID := testcore.PickRolloutSplit(t, nsName, 50)
4639
4640
mkSchedule := func() *schedulepb.Schedule {
4641
return &schedulepb.Schedule{
4642
Spec: &schedulepb.ScheduleSpec{
4643
Interval: []*schedulepb.IntervalSpec{{Interval: durationpb.New(1 * time.Hour)}},
4644
},
4645
Action: &schedulepb.ScheduleAction{
4646
Action: &schedulepb.ScheduleAction_StartWorkflow{
4647
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
4648
WorkflowId: testcore.RandomizeStr("wid"),
4649
WorkflowType: &commonpb.WorkflowType{Name: testcore.RandomizeStr("wt")},
4650
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
4651
},
4652
},
4653
},
4654
}
4655
}
4656
4657
for _, sid := range []string{chasmSID, v1SID} {
4658
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
4659
Namespace: nsName,
4660
ScheduleId: sid,
4661
Schedule: mkSchedule(),
4662
Identity: testcore.RandomizeStr("identity"),
4663
RequestId: uuid.NewString(),
4664
})
4665
require.NoError(t, err)
4666
}
4667
4668
// A direct CHASM DescribeSchedule succeeds for CHASM-backed schedules and
4669
// returns NotFound for V1-backed schedules (whose CHASM key is a sentinel).
4670
describeOnCHASM := func(sid string) error {
4671
_, err := s.GetTestCluster().SchedulerClient().DescribeSchedule(ctx, &schedulerpb.DescribeScheduleRequest{
4672
NamespaceId: nsID,
4673
FrontendRequest: &workflowservice.DescribeScheduleRequest{Namespace: nsName, ScheduleId: sid},
4674
})
4675
return err
4676
}
4677
4678
require.Eventually(t, func() bool { return describeOnCHASM(chasmSID) == nil }, 15*time.Second, 250*time.Millisecond,
4679
"schedule %q bucketed into CHASM should be describable via the CHASM handler", chasmSID)
4680
4681
var notFoundErr *serviceerror.NotFound
4682
require.ErrorAs(t, describeOnCHASM(v1SID), ¬FoundErr,
4683
"schedule %q bucketed into V1 should not be present on the CHASM handler", v1SID)
4684
4685
// Both schedules must remain describable through the public frontend
4686
// regardless of which stack they live on — the V1 schedule in particular
4687
// must round-trip through the frontend's fallback path. The V1 workflow
4688
// takes a moment to be queryable after creation, so poll.
4689
for _, sid := range []string{chasmSID, v1SID} {
4690
require.Eventually(t, func() bool {
4691
_, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4692
Namespace: nsName,
4693
ScheduleId: sid,
4694
})
4695
return err == nil
4696
}, 15*time.Second, 250*time.Millisecond, "frontend DescribeSchedule should succeed for %q", sid)
4697
}
4698
}
4699
4700
// scheduleClosesCase parameterizes the testScheduleClosesFromIdle matrix.
4701
type scheduleClosesCase struct {
4702
name string
4703
prefix string
4704
state *schedulepb.ScheduleState
4705
expectedRuns int32
4706
4707
// buildSpec receives the current time at the moment the schedule is created
4708
// so calendar/end-time-relative specs remain in the future even when test env
4709
// spinup is slow .
4710
buildSpec func(now time.Time) *schedulepb.ScheduleSpec
4711
4712
// strictRunCount asserts that runs == expectedRuns at the end (used for
4713
// LimitedActions to verify the budget is not exceeded). For calendar/end-time
4714
// variants, the spec naturally bounds runs but a stray tick is possible
4715
// before close, so we only check >=.
4716
strictRunCount bool
4717
}
4718
4719
// testScheduleClosesFromIdle runs the idle-close matrix as parallel subtests.
4720
func testScheduleClosesFromIdle(t *testing.T, newContext contextFactory) {
4721
cases := []scheduleClosesCase{
4722
{
4723
name: "SingleDate",
4724
prefix: "sched-single-date-closes",
4725
expectedRuns: 1,
4726
buildSpec: func(now time.Time) *schedulepb.ScheduleSpec {
4727
return &schedulepb.ScheduleSpec{
4728
Calendar: []*schedulepb.CalendarSpec{calendarSpec(now.Add(5 * time.Second))},
4729
}
4730
},
4731
},
4732
{
4733
name: "MultiDate",
4734
prefix: "sched-multi-date-closes",
4735
expectedRuns: 2,
4736
buildSpec: func(now time.Time) *schedulepb.ScheduleSpec {
4737
return &schedulepb.ScheduleSpec{
4738
Calendar: []*schedulepb.CalendarSpec{
4739
calendarSpec(now.Add(5 * time.Second)),
4740
calendarSpec(now.Add(10 * time.Second)),
4741
},
4742
}
4743
},
4744
},
4745
{
4746
name: "LimitedActions",
4747
prefix: "sched-limited-actions-closes",
4748
expectedRuns: 2,
4749
buildSpec: func(_ time.Time) *schedulepb.ScheduleSpec {
4750
return intervalSpec(fastInterval)
4751
},
4752
state: &schedulepb.ScheduleState{LimitedActions: true, RemainingActions: 2},
4753
strictRunCount: true,
4754
},
4755
{
4756
name: "IntervalEndTime",
4757
prefix: "sched-interval-end-closes",
4758
expectedRuns: 1,
4759
buildSpec: func(now time.Time) *schedulepb.ScheduleSpec {
4760
spec := intervalSpec(fastInterval)
4761
spec.EndTime = timestamppb.New(now.Add(10 * time.Second))
4762
return spec
4763
},
4764
},
4765
}
4766
for _, c := range cases {
4767
t.Run(c.name, func(t *testing.T) { t.Parallel(); runScheduleClosesFromIdleCase(t, newContext, c) })
4768
}
4769
}
4770
4771
// runScheduleClosesFromIdleCase drives a schedule whose spec exhausts itself,
4772
// then asserts (1) the expected number of runs fire, (2) FutureActionTimes
4773
// drains to zero, and (3) the schedule closes after IdleTime.
4774
func runScheduleClosesFromIdleCase(t *testing.T, newContext contextFactory, c scheduleClosesCase) {
4775
s := newEnvWithIdleTime(t, shortIdleTime)
4776
4777
sid := testcore.RandomizeStr(c.prefix)
4778
wid := testcore.RandomizeStr(c.prefix + "-wf")
4779
wt := testcore.RandomizeStr(c.prefix + "-wt")
4780
4781
var runs atomic.Int32
4782
registerCountingWorkflow(s, wt, &runs)
4783
4784
ctx := newContext(s.Context())
4785
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4786
Spec: c.buildSpec(time.Now().UTC()),
4787
State: c.state,
4788
Action: startWorkflowAction(s, wid, wt),
4789
})
4790
4791
// A hard action budget must land on exactly expectedRuns; time-bounded specs
4792
// may emit a stray tick before close, so they only require the lower bound.
4793
await.RequireTruef(t, func() bool {
4794
if c.strictRunCount {
4795
return runs.Load() == c.expectedRuns
4796
}
4797
return runs.Load() >= c.expectedRuns
4798
}, awaitTimeout, pollInterval, "schedule should fire its expected actions")
4799
4800
await.RequireTruef(t, func() bool {
4801
resp, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4802
Namespace: s.Namespace().String(),
4803
ScheduleId: sid,
4804
})
4805
return descErr == nil && len(resp.Info.FutureActionTimes) == 0
4806
}, awaitTimeout, pollInterval, "schedule should drain its future action times")
4807
4808
await.RequireTruef(t, func() bool { return scheduleClosed(ctx, s, sid) },
4809
awaitTimeout, pollInterval, "schedule should idle-close after IdleTime")
4810
4811
if c.strictRunCount {
4812
require.Equal(t, c.expectedRuns, runs.Load(), "schedule must not exceed its action budget")
4813
}
4814
}
4815
4816
// testManualOnlyUnpausedClosesFromIdle verifies that an unpaused manual-only
4817
// (empty-spec) schedule closes once its IdleTime elapses (as opposed to
4818
// testPausedEmptySpecStaysOpen, where a *paused* empty-spec schedule stays
4819
// open).
4820
func testManualOnlyUnpausedClosesFromIdle(t *testing.T, newContext contextFactory) {
4821
s := newEnvWithIdleTime(t, shortIdleTime)
4822
4823
sid := testcore.RandomizeStr("sched-manual-only-unpaused")
4824
wid := testcore.RandomizeStr("sched-manual-only-wf")
4825
wt := testcore.RandomizeStr("sched-manual-only-wt")
4826
4827
var runs atomic.Int32
4828
registerCountingWorkflow(s, wt, &runs)
4829
4830
ctx := newContext(s.Context())
4831
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4832
// Empty spec: a manual-only schedule, no automated actions.
4833
Spec: &schedulepb.ScheduleSpec{},
4834
Action: startWorkflowAction(s, wid, wt),
4835
})
4836
4837
// With no spec and no manual trigger, the schedule closes once its idle window elapses.
4838
await.RequireTruef(t, func() bool { return scheduleClosed(ctx, s, sid) },
4839
awaitTimeout, pollInterval, "manual-only schedule should close after idle window")
4840
require.Zero(t, runs.Load(), "a manual-only schedule must not fire any actions on its own")
4841
}
4842
4843
// testPauseDuringIdleWindow covers setting pausing, and unpausing, while a
4844
// schedule was idling.
4845
func testPauseDuringIdleWindow(t *testing.T, newContext contextFactory) {
4846
// Deliberately longer than shortIdleTime: the pause/unpause RPCs must land
4847
// inside the idle window (before the deadline) even under slow CI.
4848
idleTime := 10 * time.Second
4849
s := newEnvWithIdleTime(t, idleTime)
4850
4851
sid := testcore.RandomizeStr("sched-pause-during-idle")
4852
wid := testcore.RandomizeStr("sched-pause-during-idle-wf")
4853
wt := testcore.RandomizeStr("sched-pause-during-idle-wt")
4854
4855
var runs atomic.Int32
4856
registerCountingWorkflow(s, wt, &runs)
4857
4858
ctx := newContext(s.Context())
4859
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4860
Spec: intervalSpec(fastInterval),
4861
State: &schedulepb.ScheduleState{LimitedActions: true, RemainingActions: 1},
4862
Action: startWorkflowAction(s, wid, wt),
4863
})
4864
4865
// The single allowed action fires, exhausting the budget and arming idle.
4866
await.RequireTruef(t, func() bool { return runs.Load() == 1 },
4867
awaitTimeout, pollInterval, "the one allowed action must fire before pausing")
4868
4869
patchSchedule(ctx, t, s, sid, &schedulepb.SchedulePatch{Pause: "pause-during-idle"})
4870
4871
// Paused must hold the schedule open past the original idle deadline.
4872
require.Never(t, func() bool { return scheduleClosed(ctx, s, sid) },
4873
idleTime*2, pollInterval, "paused schedule must not close past original idle deadline")
4874
4875
// Unpause: the Generator re-arms idle and the schedule finally closes.
4876
patchSchedule(ctx, t, s, sid, &schedulepb.SchedulePatch{Unpause: "resume-after-idle"})
4877
await.RequireTruef(t, func() bool { return scheduleClosed(ctx, s, sid) },
4878
awaitTimeout, pollInterval, "schedule must close after unpause via re-armed idle task")
4879
require.Equal(t, int32(1), runs.Load(), "no extra actions should fire across pause/unpause")
4880
}
4881
4882
// testBackfillBlocksIdleClose verifies that a schedule with no remaining
4883
// automated actions and a pending backfill is not closed by the idle path
4884
// until it drains.
4885
func testBackfillBlocksIdleClose(t *testing.T, newContext contextFactory) {
4886
s := newEnvWithIdleTime(t, shortIdleTime)
4887
4888
sid := testcore.RandomizeStr("sched-backfill-blocks-idle")
4889
wid := testcore.RandomizeStr("sched-backfill-blocks-idle-wf")
4890
wt := testcore.RandomizeStr("sched-backfill-blocks-idle-wt")
4891
4892
var runs atomic.Int32
4893
registerCountingWorkflow(s, wt, &runs)
4894
4895
ctx := newContext(s.Context())
4896
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4897
Spec: intervalSpec(fastInterval),
4898
State: &schedulepb.ScheduleState{
4899
LimitedActions: true,
4900
RemainingActions: 1,
4901
},
4902
Action: startWorkflowAction(s, wid, wt),
4903
})
4904
4905
// The single allowed automated action fires, leaving the scheduler heading to idle.
4906
await.RequireTruef(t, func() bool { return runs.Load() == 1 }, awaitTimeout, pollInterval,
4907
"the single allowed automated action should have fired")
4908
4909
// BUFFER_ALL is used to force each to run sequentially (versus in parallel with
4910
// ALLOW_ALL), which is a better test to show the idle time is pushed back.
4911
now := time.Now().UTC()
4912
patchSchedule(ctx, t, s, sid, backfillPatch(now.Add(-10*time.Second), now, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL))
4913
4914
// Backfill fires despite the scheduler heading to idle; tick count in the 5s
4915
// window varies with boundary alignment, so assert the lower bound.
4916
await.RequireTruef(t, func() bool { return runs.Load() >= 10 },
4917
awaitTimeout, pollInterval,
4918
"backfill should fire actions even though the scheduler was heading to idle")
4919
4920
await.RequireTruef(t, func() bool { return scheduleClosed(ctx, s, sid) },
4921
awaitTimeout, pollInterval,
4922
"scheduler should close from idle once the backfill drains and IdleTime elapses")
4923
}
4924
4925
// testMultiRangeBackfillCountedExactlyOnce asserts that the `ActionCount` is
4926
// correctly counted when multiple backfillers are concurrently running.
4927
func testMultiRangeBackfillCountedExactlyOnce(t *testing.T, newContext contextFactory) {
4928
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4929
4930
sid := testcore.RandomizeStr("sched-multi-range-backfill")
4931
wid := testcore.RandomizeStr("sched-multi-range-backfill-wf")
4932
wt := testcore.RandomizeStr("sched-multi-range-backfill-wt")
4933
4934
s.SdkWorker().RegisterWorkflowWithOptions(
4935
func(ctx workflow.Context) error { return nil },
4936
workflow.RegisterOptions{Name: wt},
4937
)
4938
4939
ctx := newContext(s.Context())
4940
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4941
// 1m interval; each 2m backfill range below covers a couple of ticks.
4942
Spec: intervalSpec(time.Minute),
4943
State: &schedulepb.ScheduleState{Paused: true},
4944
Action: startWorkflowAction(s, wid, wt),
4945
})
4946
4947
now := time.Now().UTC()
4948
threeYearsAgo := now.Add(-3 * 365 * 24 * time.Hour).Truncate(time.Minute)
4949
thirtyMinutesAgo := now.Add(-30 * time.Minute).Truncate(time.Minute)
4950
patchSchedule(ctx, t, s, sid, &schedulepb.SchedulePatch{
4951
BackfillRequest: []*schedulepb.BackfillRequest{
4952
{
4953
StartTime: timestamppb.New(threeYearsAgo.Add(-2 * time.Minute)),
4954
EndTime: timestamppb.New(threeYearsAgo),
4955
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
4956
},
4957
{
4958
StartTime: timestamppb.New(thirtyMinutesAgo.Add(-2 * time.Minute)),
4959
EndTime: timestamppb.New(thirtyMinutesAgo),
4960
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL,
4961
},
4962
},
4963
})
4964
4965
await.RequireTruef(t, func() bool {
4966
desc, descErr := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
4967
Namespace: s.Namespace().String(),
4968
ScheduleId: sid,
4969
})
4970
if descErr != nil {
4971
return false
4972
}
4973
return desc.Info.ActionCount == 6 && len(desc.Info.RunningWorkflows) == 0
4974
}, awaitTimeout, pollInterval,
4975
"backfill should fire 6 actions and complete on a paused schedule")
4976
}
4977
4978
// testBackfillOnPausedSchedule verifies that a paused schedule still processes
4979
// a backfill request to completion, even though the schedule otherwise has no
4980
// automated actions running.
4981
func testBackfillOnPausedSchedule(t *testing.T, newContext contextFactory) {
4982
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
4983
4984
sid := testcore.RandomizeStr("sched-backfill-paused")
4985
wid := testcore.RandomizeStr("sched-backfill-paused-wf")
4986
wt := testcore.RandomizeStr("sched-backfill-paused-wt")
4987
4988
var runs atomic.Int32
4989
registerCountingWorkflow(s, wt, &runs)
4990
4991
ctx := newContext(s.Context())
4992
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
4993
Spec: intervalSpec(fastInterval),
4994
State: &schedulepb.ScheduleState{Paused: true},
4995
Action: startWorkflowAction(s, wid, wt),
4996
})
4997
4998
// Paused suppresses firing even though the 1s interval would otherwise tick.
4999
require.Never(t, func() bool { return runs.Load() > 0 },
5000
neverWindow, pollInterval,
5001
"paused schedule must not fire automated actions")
5002
5003
now := time.Now().UTC()
5004
patchSchedule(ctx, t, s, sid, backfillPatch(now.Add(-5*time.Second), now, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL))
5005
5006
await.RequireTruef(t, func() bool { return runs.Load() >= 5 },
5007
awaitTimeout, pollInterval,
5008
"backfill should fire actions on a paused schedule")
5009
}
5010
5011
// TestScheduleNextActionTimeVisibility asserts that the CHASM scheduler's
5012
// ScheduleNextActionTime search attribute is published to visibility and is
5013
// queryable through the frontend ListSchedules API.
5014
func TestScheduleNextActionTimeVisibility(t *testing.T) {
5015
opts := scheduleCommonOpts(t)
5016
s := newScheduleEnv(t, opts...)
5017
5018
v2Sid := testcore.RandomizeStr("sched-next-action-v2")
5019
wid := testcore.RandomizeStr("sched-next-action-wf")
5020
wt := testcore.RandomizeStr("sched-next-action-wt")
5021
5022
// Register a no-op workflow type so a stray fire doesn't generate worker
5023
// noise. The schedule is never paused: we want its next action time to stay
5024
// populated so it remains queryable.
5025
s.SdkWorker().RegisterWorkflowWithOptions(
5026
func(ctx workflow.Context) error { return nil },
5027
workflow.RegisterOptions{Name: wt},
5028
)
5029
5030
mkSchedule := func() *schedulepb.Schedule {
5031
return &schedulepb.Schedule{
5032
Spec: &schedulepb.ScheduleSpec{
5033
Interval: []*schedulepb.IntervalSpec{
5034
{Interval: durationpb.New(1 * time.Hour), Phase: durationpb.New(23 * time.Minute)},
5035
},
5036
},
5037
Action: &schedulepb.ScheduleAction{
5038
Action: &schedulepb.ScheduleAction_StartWorkflow{
5039
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
5040
WorkflowId: wid,
5041
WorkflowType: &commonpb.WorkflowType{Name: wt},
5042
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
5043
},
5044
},
5045
},
5046
}
5047
}
5048
5049
newContext := chasmContextFactory
5050
v2Ctx := newContext(s.Context())
5051
createTime := time.Now()
5052
5053
_, err := s.FrontendClient().CreateSchedule(v2Ctx, &workflowservice.CreateScheduleRequest{
5054
Namespace: s.Namespace().String(),
5055
ScheduleId: v2Sid,
5056
Schedule: mkSchedule(),
5057
Identity: "test",
5058
RequestId: uuid.NewString(),
5059
})
5060
s.NoError(err)
5061
5062
// The CHASM scheduler publishes its next action time to visibility as the
5063
// ScheduleNextActionTime search attribute. The schedule fires on an hourly
5064
// interval, so its next action time is always in the future; a query for
5065
// ScheduleNextActionTime > createTime must therefore eventually return this
5066
// schedule. (The SA is indexed/queryable but not surfaced on the list entry,
5067
// so we assert via the query rather than by reading the entry's SAs.)
5068
query := fmt.Sprintf(`%s > "%s"`,
5069
chasmscheduler.ScheduleNextActionTimeName,
5070
createTime.UTC().Format(time.RFC3339Nano),
5071
)
5072
5073
require.Eventually(t, func() bool {
5074
listResp, err := s.FrontendClient().ListSchedules(v2Ctx, &workflowservice.ListSchedulesRequest{
5075
Namespace: s.Namespace().String(),
5076
MaximumPageSize: 100,
5077
Query: query,
5078
})
5079
if err != nil {
5080
return false
5081
}
5082
for _, ent := range listResp.Schedules {
5083
if ent.ScheduleId == v2Sid {
5084
return true
5085
}
5086
}
5087
return false
5088
}, 15*time.Second, 1*time.Second,
5089
"schedule %q must be returned by query %q (next action time published to visibility and in the future)",
5090
v2Sid, query)
5091
}
5092
5093
// TestMirroredIncludeExcludeSpec sets identical interval and exclusion
5094
// specifications that match every 1s, effectively cancelling each other out.
5095
func TestMirroredIncludeExcludeSpec(t *testing.T) {
5096
// A tiny compute bound trips the mirrored spec near-instantly; the default (~1.2M candidate
5097
// scans per GetNextTime) makes this test burn seconds of CPU on every scheduler code path.
5098
opts := append(scheduleCommonOpts(t), testcore.WithDynamicConfig(dynamicconfig.SchedulerSpecMaxIterations, 1000))
5099
s := testcore.NewEnv(t, opts...)
5100
5101
sid := testcore.RandomizeStr("sched-cancelling-spec")
5102
wid := testcore.RandomizeStr("sched-cancelling-spec-wf")
5103
wt := testcore.RandomizeStr("sched-cancelling-spec-wt")
5104
5105
everySecond := &schedulepb.CalendarSpec{Second: "*", Minute: "*", Hour: "*"}
5106
5107
ctx, cancel := context.WithTimeout(chasmContextFactory(s.Context()), 10*time.Second)
5108
defer cancel()
5109
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
5110
Spec: &schedulepb.ScheduleSpec{
5111
Calendar: []*schedulepb.CalendarSpec{everySecond},
5112
ExcludeCalendar: []*schedulepb.CalendarSpec{everySecond},
5113
},
5114
Action: startWorkflowAction(s, wid, wt),
5115
})
5116
5117
// ListMatchingTimes must surface the compute limit as an error rather than hang.
5118
_, lmErr := s.FrontendClient().ListScheduleMatchingTimes(ctx, &workflowservice.ListScheduleMatchingTimesRequest{
5119
Namespace: s.Namespace().String(),
5120
ScheduleId: sid,
5121
StartTime: timestamppb.New(time.Now().UTC()),
5122
EndTime: timestamppb.New(time.Now().UTC().Add(time.Hour)),
5123
})
5124
require.Error(t, lmErr, "ListMatchingTimes should error for a mirrored include/exclude spec")
5125
}
5126
5127
// TestMirroredIncludeExcludeSpecOnUpdate is like TestMirroredIncludeExcludeSpec but reaches the
5128
// mirrored spec via UpdateSchedule, exercising the spec-recompile path on an existing schedule.
5129
func TestMirroredIncludeExcludeSpecOnUpdate(t *testing.T) {
5130
// A tiny compute bound trips the mirrored spec near-instantly (see TestMirroredIncludeExcludeSpec).
5131
opts := append(scheduleCommonOpts(t), testcore.WithDynamicConfig(dynamicconfig.SchedulerSpecMaxIterations, 1000))
5132
s := testcore.NewEnv(t, opts...)
5133
5134
sid := testcore.RandomizeStr("sched-cancelling-update")
5135
wid := testcore.RandomizeStr("sched-cancelling-update-wf")
5136
wt := testcore.RandomizeStr("sched-cancelling-update-wt")
5137
5138
ctx := chasmContextFactory(s.Context())
5139
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
5140
Spec: intervalSpec(1 * time.Hour),
5141
Action: startWorkflowAction(s, wid, wt),
5142
})
5143
5144
everySecond := &schedulepb.CalendarSpec{Second: "*", Minute: "*", Hour: "*"}
5145
updateCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
5146
defer cancel()
5147
_, err := s.FrontendClient().UpdateSchedule(updateCtx, &workflowservice.UpdateScheduleRequest{
5148
Namespace: s.Namespace().String(),
5149
ScheduleId: sid,
5150
Schedule: &schedulepb.Schedule{
5151
Spec: &schedulepb.ScheduleSpec{
5152
Calendar: []*schedulepb.CalendarSpec{everySecond},
5153
ExcludeCalendar: []*schedulepb.CalendarSpec{everySecond},
5154
},
5155
Action: startWorkflowAction(s, wid, wt),
5156
},
5157
Identity: "test",
5158
RequestId: uuid.NewString(),
5159
})
5160
require.NoError(t, err, "UpdateSchedule to a mirrored include/exclude spec should not hang")
5161
5162
// Once the mirrored spec is applied, ListMatchingTimes must surface the compute limit.
5163
await.RequireTruef(t, func() bool {
5164
_, lmErr := s.FrontendClient().ListScheduleMatchingTimes(ctx, &workflowservice.ListScheduleMatchingTimesRequest{
5165
Namespace: s.Namespace().String(),
5166
ScheduleId: sid,
5167
StartTime: timestamppb.New(time.Now().UTC()),
5168
EndTime: timestamppb.New(time.Now().UTC().Add(time.Hour)),
5169
})
5170
return lmErr != nil
5171
}, awaitTimeout, pollInterval, "ListMatchingTimes should error once the mirrored spec is applied")
5172
}
5173
5174
// TestScheduleFarFutureActionTimes verifies that a schedule firing far beyond the compute
5175
// horizon (an interval 10x the horizon) still fills FutureActionTimes, since each far-future
5176
// time is found in a single interval step rather than by scanning.
5177
func TestScheduleFarFutureActionTimes(t *testing.T) {
5178
s := testcore.NewEnv(t, scheduleCommonOpts(t)...)
5179
5180
sid := testcore.RandomizeStr("sched-far-future")
5181
wid := testcore.RandomizeStr("sched-far-future-wf")
5182
wt := testcore.RandomizeStr("sched-far-future-wt")
5183
5184
s.SdkWorker().RegisterWorkflowWithOptions(
5185
func(ctx workflow.Context) error { return nil },
5186
workflow.RegisterOptions{Name: wt},
5187
)
5188
5189
warn := time.Duration(scheduler.DefaultWarnIterations) * time.Second
5190
interval := 10 * warn
5191
5192
ctx := chasmContextFactory(s.Context())
5193
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
5194
Spec: intervalSpec(interval),
5195
Action: startWorkflowAction(s, wid, wt),
5196
})
5197
5198
var future []*timestamppb.Timestamp
5199
await.RequireTruef(t, func() bool {
5200
resp, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
5201
Namespace: s.Namespace().String(),
5202
ScheduleId: sid,
5203
})
5204
if err != nil || len(resp.GetInfo().GetFutureActionTimes()) < 10 {
5205
return false
5206
}
5207
future = resp.GetInfo().GetFutureActionTimes()
5208
return true
5209
}, awaitTimeout, pollInterval, "FutureActionTimes should fill to 10 far-future entries")
5210
5211
for i := 1; i < len(future); i++ {
5212
require.Equal(t, interval, future[i].AsTime().Sub(future[i-1].AsTime()),
5213
"consecutive future action times should be exactly one interval apart")
5214
}
5215
require.Greater(t, future[len(future)-1].AsTime().Sub(time.Now().UTC()), warn,
5216
"future action times should extend well past the two-week compute horizon")
5217
5218
base := time.Now().UTC()
5219
resp, err := s.FrontendClient().ListScheduleMatchingTimes(ctx, &workflowservice.ListScheduleMatchingTimesRequest{
5220
Namespace: s.Namespace().String(),
5221
ScheduleId: sid,
5222
StartTime: timestamppb.New(base),
5223
EndTime: timestamppb.New(base.Add(11 * interval)),
5224
})
5225
require.NoError(t, err)
5226
require.GreaterOrEqual(t, len(resp.GetStartTime()), 10,
5227
"ListMatchingTimes should enumerate the far-future series")
5228
}
5229
5230
// TestScheduleManyCalendars verifies that a spec with 50 calendars and 50 excludes still
5231
// evaluates cheaply, so the schedule keeps dispatching actions while RecentActions and
5232
// FutureActionTimes keep updating.
5233
func TestScheduleManyCalendars(t *testing.T) {
5234
s := testcore.NewEnv(t, scheduleCommonOpts(t)...)
5235
5236
sid := testcore.RandomizeStr("sched-many-calendars")
5237
wid := testcore.RandomizeStr("sched-many-calendars-wf")
5238
wt := testcore.RandomizeStr("sched-many-calendars-wt")
5239
5240
var runs atomic.Int32
5241
registerCountingWorkflow(s, wt, &runs)
5242
5243
// Calendars match seconds 0..49 every minute so the schedule fires often. The excludes cancel
5244
// every fire in one upcoming minute (computed from now) so they actually take effect, while
5245
// the schedule keeps firing every other minute.
5246
const n = 50
5247
excludeMinute := time.Now().UTC().Truncate(time.Minute).Add(2 * time.Minute)
5248
var calendars, excludes []*schedulepb.CalendarSpec
5249
for i := range n {
5250
calendars = append(calendars, &schedulepb.CalendarSpec{
5251
Second: fmt.Sprintf("%d", i), Minute: "*", Hour: "*",
5252
})
5253
excludes = append(excludes, calendarSpec(excludeMinute.Add(time.Duration(i)*time.Second)))
5254
}
5255
5256
ctx := chasmContextFactory(s.Context())
5257
createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
5258
Spec: &schedulepb.ScheduleSpec{
5259
Calendar: calendars,
5260
ExcludeCalendar: excludes,
5261
},
5262
Action: startWorkflowAction(s, wid, wt),
5263
})
5264
5265
await.RequireTruef(t, func() bool { return runs.Load() >= 3 },
5266
awaitTimeout, pollInterval, "schedule with many calendars should keep dispatching actions")
5267
5268
describe := func() *schedulepb.ScheduleInfo {
5269
resp, err := s.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{
5270
Namespace: s.Namespace().String(),
5271
ScheduleId: sid,
5272
})
5273
require.NoError(t, err)
5274
return resp.GetInfo()
5275
}
5276
5277
var baseRecent, baseFuture time.Time
5278
await.RequireTruef(t, func() bool {
5279
info := describe()
5280
if len(info.GetRecentActions()) == 0 || len(info.GetFutureActionTimes()) == 0 {
5281
return false
5282
}
5283
baseRecent = info.GetRecentActions()[len(info.GetRecentActions())-1].GetActualTime().AsTime()
5284
baseFuture = info.GetFutureActionTimes()[0].AsTime()
5285
return true
5286
}, awaitTimeout, pollInterval, "RecentActions and FutureActionTimes should be populated")
5287
5288
await.RequireTruef(t, func() bool {
5289
info := describe()
5290
if len(info.GetRecentActions()) == 0 || len(info.GetFutureActionTimes()) == 0 {
5291
return false
5292
}
5293
newerRecent := info.GetRecentActions()[len(info.GetRecentActions())-1].GetActualTime().AsTime().After(baseRecent)
5294
advancedFuture := info.GetFutureActionTimes()[0].AsTime().After(baseFuture)
5295
return newerRecent && advancedFuture
5296
}, awaitTimeout, pollInterval, "RecentActions and FutureActionTimes should keep updating as actions fire")
5297
5298
// The excludes cancel every fire in excludeMinute, so ListMatchingTimes over it is empty
5299
// (the schedule keeps firing in other minutes, asserted above).
5300
resp, err := s.FrontendClient().ListScheduleMatchingTimes(ctx, &workflowservice.ListScheduleMatchingTimesRequest{
5301
Namespace: s.Namespace().String(),
5302
ScheduleId: sid,
5303
StartTime: timestamppb.New(excludeMinute.Add(-time.Second)),
5304
EndTime: timestamppb.New(excludeMinute.Add(50 * time.Second)),
5305
})
5306
require.NoError(t, err)
5307
require.Empty(t, resp.GetStartTime(), "the excluded minute must have no matches")
5308
}
5309
5310
// TestScheduleCountsVisibility asserts that the CHASM scheduler's
5311
// ScheduleRunningWorkflowCount and ScheduleBufferedStartsCount search attributes
5312
// are published to visibility and queryable through the frontend ListSchedules
5313
// API. A schedule that fires every second under BUFFER_ONE, started with a
5314
// workflow that blocks, settles into one running workflow with one fire buffered
5315
// behind it. The counts aren't surfaced on the list entry, so we assert via the
5316
// query rather than by reading the entry's SAs.
5317
func TestScheduleCountsVisibility(t *testing.T) {
5318
s := newScheduleEnv(t, scheduleCommonOpts(t)...)
5319
newContext := chasmContextFactory
5320
5321
sid := testcore.RandomizeStr("sched-counts-v2")
5322
wid := testcore.RandomizeStr("sched-counts-wf")
5323
wt := testcore.RandomizeStr("sched-counts-wt")
5324
5325
// A workflow that holds open so a fire stays running while later fires buffer.
5326
s.SdkWorker().RegisterWorkflowWithOptions(
5327
func(ctx workflow.Context) error {
5328
return workflow.Sleep(ctx, time.Hour)
5329
},
5330
workflow.RegisterOptions{Name: wt},
5331
)
5332
5333
schedule := &schedulepb.Schedule{
5334
Spec: &schedulepb.ScheduleSpec{
5335
Interval: []*schedulepb.IntervalSpec{
5336
{Interval: durationpb.New(1 * time.Second)},
5337
},
5338
},
5339
Action: &schedulepb.ScheduleAction{
5340
Action: &schedulepb.ScheduleAction_StartWorkflow{
5341
StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
5342
WorkflowId: wid,
5343
WorkflowType: &commonpb.WorkflowType{Name: wt},
5344
TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
5345
},
5346
},
5347
},
5348
Policies: &schedulepb.SchedulePolicies{
5349
OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE,
5350
},
5351
}
5352
5353
ctx := newContext(s.Context())
5354
_, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{
5355
Namespace: s.Namespace().String(),
5356
ScheduleId: sid,
5357
Schedule: schedule,
5358
Identity: "test",
5359
RequestId: uuid.NewString(),
5360
})
5361
require.NoError(t, err)
5362
5363
// matchesQuery reports whether the schedule is returned when filtering on the
5364
// given search-attribute query.
5365
matchesQuery := func(query string) bool {
5366
listResp, listErr := s.FrontendClient().ListSchedules(newContext(s.Context()), &workflowservice.ListSchedulesRequest{
5367
Namespace: s.Namespace().String(),
5368
MaximumPageSize: 5,
5369
Query: query,
5370
})
5371
if listErr != nil {
5372
return false
5373
}
5374
for _, ent := range listResp.Schedules {
5375
if ent.ScheduleId == sid {
5376
return true
5377
}
5378
}
5379
return false
5380
}
5381
5382
// Starting the workflow and buffering the next fire takes a few seconds on top
5383
// of visibility propagation, so allow 30s for each count to become queryable.
5384
require.Eventually(t, func() bool {
5385
return matchesQuery(fmt.Sprintf("%s >= 1", chasmscheduler.ScheduleRunningWorkflowCountName))
5386
}, 30*time.Second, 500*time.Millisecond,
5387
"schedule must be queryable by ScheduleRunningWorkflowCount >= 1")
5388
5389
require.Eventually(t, func() bool {
5390
return matchesQuery(fmt.Sprintf("%s >= 1", chasmscheduler.ScheduleBufferedStartsCountName))
5391
}, 30*time.Second, 500*time.Millisecond,
5392
"schedule must be queryable by ScheduleBufferedStartsCount >= 1")
5393
}