Vihren Atlas

scheduler.go ×19

Code frontier

unlabeled · c_9f03d7b21370

146 tests · 4987 LOC · 168 files · introduces 0 tests · 240 LOC · 9 files

Introduces — evidence that enters the hierarchy at this concept

Code
55 ranges240 lines · 9 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
1131 ranges4987 lines · 168 files · Browse complete extent
All tests (intent)
146 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

9 files ranked by introduced lines: 240 introduced LOC across 55 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/chasm/lib/scheduler/scheduler.go 88 introduced LOC · 19 ranges

Open complete file

206
207 // setNullableFields sets fields that are nullable in API requests.
208 > func (s *Scheduler) setNullableFields() { scheduler.go
209 > if s.Schedule.Policies == nil {
210 s.Schedule.Policies = &schedulepb.SchedulePolicies{}
211 }
212 > if s.Schedule.State == nil { scheduler.go
213 s.Schedule.State = &schedulepb.ScheduleState{}
214 }
390 // decremented when an action can be taken. When decrement is false, no state
391 // is mutated.
392 > func (s *Scheduler) useScheduledAction(decrement bool) bool { scheduler.go
393 > scheduleState := s.Schedule.GetState()
394 >
395 > // If paused, don't do anything.
396 > if scheduleState.Paused {
397 return false
398 }
399
400 // If unlimited actions, allow.
401 > if !scheduleState.LimitedActions { scheduler.go
402 > return true
403 > }
404
405 // Otherwise check and decrement limit.
420 }
421
422 > func (s *Scheduler) getCompiledSpec(specBuilder *scheduler.SpecBuilder) (*scheduler.CompiledSpec, error) { scheduler.go
423 > s.validateCachedState()
424 >
425 > // Cache compiled spec.
426 > if s.compiledSpec == nil {
427 > cspec, err := specBuilder.NewCompiledSpec(s.Schedule.Spec)
428 > if err != nil {
429 return nil, err
430 }
431 > s.compiledSpec = cspec scheduler.go
432 }
433
434 > return s.compiledSpec, nil scheduler.go
435 }
436
437 // WorkflowID returns the Workflow ID given as part of the request spec.
438 // During start generation, nominal time is suffixed to this ID.
439 > func (s *Scheduler) WorkflowID() string { scheduler.go
440 > return s.Schedule.GetAction().GetStartWorkflow().GetWorkflowId()
441 > }
442
443 > func (s *Scheduler) jitterSeed() string { scheduler.go
444 > return fmt.Sprintf("%s-%s", s.NamespaceId, s.ScheduleId)
445 > }
446
447 func (s *Scheduler) identity() string {
449 }
450
451 > func (s *Scheduler) overlapPolicy() enumspb.ScheduleOverlapPolicy { scheduler.go
452 > policy := s.Schedule.GetPolicies().GetOverlapPolicy()
453 > if policy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {
454 > policy = enumspb.SCHEDULE_OVERLAP_POLICY_SKIP
455 > }
456 > return policy
457 }
458
467 // ConflictToken doesn't match its cacheConflictToken field. Validation is only
468 // as effective as the Scheduler's backing persisted state is up-to-date.
469 > func (s *Scheduler) validateCachedState() { scheduler.go
470 > if s.cacheConflictToken != s.ConflictToken {
471 // Bust stale cached fields.
472 s.compiledSpec = nil
515 return false
516 }
517 > return s.Schedule.GetState().GetPaused() || scheduler.go
518 > s.hasMoreBackfills()
519 }
520
536 idleTime time.Duration,
537 nextWakeup time.Time,
538 > ) (time.Time, bool) { scheduler.go
539 > if idleTime == 0 ||
540 > s.isHeldOpen() ||
541 > (!nextWakeup.IsZero() && s.useScheduledAction(false)) {
542 > return time.Time{}, false
543 > }
544 return s.idleDeadline(ctx, idleTime), true
545 }
546
547 > func (s *Scheduler) hasMoreBackfills() bool { scheduler.go
548 > return len(s.Backfillers) > 0
549 > }
550
551 type schedulerActionResult struct {
988 }
989 }
990 > out := []chasm.SearchAttributeKeyValue{ scheduler.go
991 > executionStatusSearchAttribute.Value(s.executionStatus()),
992 > chasm.SearchAttributeTemporalSchedulePaused.Value(s.Schedule.GetState().GetPaused()),
993 > }
994 > if !s.Closed {
995 > if gen := s.Generator.Get(ctx); len(gen.FutureActionTimes) > 0 {
996 > out = append(out, scheduleNextActionTimeSearchAttribute.Value(gen.FutureActionTimes[0].AsTime()))
997 > }
998 > if s.IdleCloseTime != nil {
999 out = append(out, scheduleIdleCloseTimeSearchAttribute.Value(s.IdleCloseTime.AsTime()))
1000 }
1001
1002 > invoker := s.Invoker.Get(ctx) scheduler.go
1003 > runningWorkflowCount := int64(len(invoker.runningWorkflowExecutions()))
1004 > bufferedStartsCount := int64(len(invoker.GetBufferedStarts()) - len(invoker.recentActions()))
1005 >
1006 > // Emitted even when zero so that exact and range queries both work.
1007 > out = append(out,
1008 > scheduleRunningWorkflowCountSearchAttribute.Value(runningWorkflowCount),
1009 > scheduleBufferedStartsCountSearchAttribute.Value(bufferedStartsCount),
1010 > )
1011 }
1012 > return out scheduler.go
1013 }
1014
1020 return nil
1021 }
1022 > return s.ListInfo(ctx) scheduler.go
1023 }
1024
1027 func (s *Scheduler) ListInfo(
1028 ctx chasm.Context,
1029 > ) *schedulepb.ScheduleListInfo { scheduler.go
1030 > spec := common.CloneProto(s.Schedule.Spec)
1031 >
1032 > // Clear fields that are too large/not useful for the list view.
1033 > spec.TimezoneData = nil
1034 >
1035 > // Limit the number of specs and exclusions stored on the memo.
1036 > spec.ExcludeStructuredCalendar = util.SliceHead(spec.ExcludeStructuredCalendar, listInfoSpecFieldLimit)
1037 > spec.Interval = util.SliceHead(spec.Interval, listInfoSpecFieldLimit)
1038 > spec.StructuredCalendar = util.SliceHead(spec.StructuredCalendar, listInfoSpecFieldLimit)
1039 >
1040 > generator := s.Generator.Get(ctx)
1041 > invoker := s.Invoker.Get(ctx)
1042 >
1043 > return &schedulepb.ScheduleListInfo{
1044 > Spec: spec,
1045 > WorkflowType: s.Schedule.Action.GetStartWorkflow().GetWorkflowType(),
1046 > Notes: s.Schedule.State.Notes,
1047 > Paused: s.Schedule.State.Paused,
1048 > RecentActions: invoker.recentActions(),
1049 > FutureActionTimes: generator.FutureActionTimes,
1050 > }
1051 > }
1052
1053 // startWorkflowSearchAttributes returns the search attributes to be applied to
go.temporal.io/server/chasm/lib/scheduler/generator_tasks.go 65 introduced LOC · 12 ranges

Open complete file

53 _ chasm.TaskAttributes,
54 _ *schedulerpb.GeneratorTask,
55 > ) error { generator_tasks.go
56 > scheduler := generator.Scheduler.Get(ctx)
57 > logger := newTaggedLogger(g.baseLogger, scheduler)
58 > metricsHandler := newTaggedMetricsHandler(g.metricsHandler, scheduler)
59 > invoker := scheduler.Invoker.Get(ctx)
60 >
61 > now := ctx.Now(generator)
62 >
63 > generator.getOrCreateEventLog(ctx).LogEvent(ctx, "generatorTask executed")
64 >
65 > // If we have no last processed time, this is a new schedule.
66 > if generator.LastProcessedTime == nil {
67 createdAt := timestamppb.New(now)
68 generator.LastProcessedTime = createdAt
74 // If the high water mark is earlier than when a schedule was updated, we must skip any actions that hadn't
75 // yet been processed.
76 > if scheduler.Info.GetUpdateTime().AsTime().After(generator.LastProcessedTime.AsTime()) { generator_tasks.go
77 generator.LastProcessedTime = scheduler.Info.GetUpdateTime()
78 }
79
80 // Process time range between last high water mark and system time.
81 > t1 := generator.LastProcessedTime.AsTime() generator_tasks.go
82 > t2 := now.UTC()
83 > if t2.Before(t1) {
84 logger.Error("time went backwards",
85 tag.Stringer("time", t1),
88 }
89
90 > tweakables := g.config.Tweakables(scheduler.Namespace) generator_tasks.go
91 > var limit *int
92 > if tweakables.MaxBufferSize > 0 {
93 > remaining := tweakables.MaxBufferSize - len(invoker.GetBufferedStarts())
94 > limit = &remaining
95 > }
96
97 // Generate BufferedStarts and determine the next HWM. Actions are skipped when
98 // they can't be taken (paused, or limited and without any remaining actions),
99 // and dropped when the buffer is full.
100 > result, err := g.SpecProcessor.ProcessTimeRange( generator_tasks.go
101 > scheduler,
102 > t1, t2,
103 > scheduler.overlapPolicy(),
104 > scheduler.WorkflowID(),
105 > "",
106 > false,
107 > limit,
108 > )
109 > if err != nil {
110 // An error here should be impossible, send to the DLQ.
111 return queueerrors.NewUnprocessableTaskError(
113 }
114
115 > if result.DroppedCount > 0 { generator_tasks.go
116 // Only system log on the first drop, as it's likely that a case that overruns
117 // will continue to overrun.
130 // to paused schedules vs. real work. Each fire while paused advances the
131 // HWM without buffering anything.
132 > metricsHandler.Counter(metrics.ScheduleGeneratorTicks.Name()).Record(1) generator_tasks.go
133 > if scheduler.Schedule.State.Paused {
134 metricsHandler.Counter(metrics.ScheduleGeneratorPausedTicks.Name()).Record(1)
135 }
136
137 // Enqueue newly-generated buffered starts.
138 > if len(result.BufferedStarts) > 0 { generator_tasks.go
139 invoker.EnqueueBufferedStarts(ctx, result.BufferedStarts)
140 }
141
142 // Write the new high water mark and future action times.
143 > generator.LastProcessedTime = timestamppb.New(result.LastActionTime) generator_tasks.go
144 > generator.UpdateFutureActionTimes(ctx, g.specBuilder)
145 >
146 > // Schedule the next timer task. Three outcomes are possible:
147 > // - isIdle: the schedule is done; arm the idle task to close it.
148 > // - NextWakeupTime is set: arm the next generator tick.
149 > // - Neither: Hold open without a task. This requires both that
150 > // isHeldOpen is true (paused or backfill pending) AND that no spec
151 > // wakeup is available, e.g. a paused manual-only schedule. IdleTime=0
152 > // also lands here. An external trigger (Patch.Unpause, Update, or a
153 > // BackfillerTask's completion-time Generate call) revives us.
154 > idleTimeTotal := tweakables.IdleTime
155 > idleExpiration, isIdle := scheduler.getIdleExpiration(ctx, idleTimeTotal, result.NextWakeupTime)
156 > if isIdle {
157 // Schedule is complete, no need for another buffer task. We keep the schedule's
158 // backing mutable state explicitly open for the idle period, during which the
176 // Not idle: the schedule has work again (or is being held open), so it's
177 // no longer pending an idle close.
178 > scheduler.IdleCloseTime = nil generator_tasks.go
179 >
180 > if !result.NextWakeupTime.IsZero() {
181 > // Keep the generator task perpetually scheduled. When paused, the next
182 > // fire will simply advance the HWM without appending actions (handled in
183 > // ProcessTimeRange).
184 > generator.scheduleTask(ctx, result.NextWakeupTime)
185 > } else {
186 // Hold open without a task: see the comment block above.
187 metricsHandler.Counter(metrics.SchedulerGeneratorLoopCompleted.Name()).Record(1)
188 }
189
190 > return nil generator_tasks.go
191 }
192
206 attrs chasm.TaskInvocation,
207 _ *schedulerpb.GeneratorTask,
208 > ) (bool, error) { generator_tasks.go
209 > return validateTaskHighWaterMark(
210 > generator.GetLastProcessedTime(),
211 > attrs.ScheduledTime,
212 > )
213 > }
go.temporal.io/server/chasm/lib/scheduler/generator.go 40 introduced LOC · 12 ranges

Open complete file

36 }
37
38 > func newGeneratorWithState(ctx chasm.MutableContext, state *schedulerpb.GeneratorState) *Generator { generator.go
39 > generator := &Generator{
40 > GeneratorState: state,
41 > EventLog: chasm.NewComponentField(ctx, NewEventLog(ctx)),
42 > }
43 > return generator
44 > }
45
46 // Generate immediately kicks off a new GeneratorTask. Used after updating the
47 // schedule specification.
48 > func (g *Generator) Generate(ctx chasm.MutableContext) { generator.go
49 > g.scheduleTask(ctx, chasm.TaskScheduledTimeImmediate)
50 > }
51
52 // scheduleTask schedules a GeneratorTask at the given time.
53 > func (g *Generator) scheduleTask(ctx chasm.MutableContext, scheduledTime time.Time) { generator.go
54 > g.getOrCreateEventLog(ctx).LogEvent(ctx,
55 > fmt.Sprintf("scheduled generatorTask for %s", scheduledTime.Format(time.RFC3339)))
56 > ctx.AddTask(g, chasm.TaskAttributes{
57 > ScheduledTime: scheduledTime,
58 > }, &schedulerpb.GeneratorTask{})
59 > }
60
61 > func (g *Generator) LifecycleState(ctx chasm.Context) chasm.LifecycleState { generator.go
62 > return chasm.LifecycleStateRunning
63 > }
64
65 // UpdateFutureActionTimes computes and stores the next scheduled action times.
68 ctx chasm.MutableContext,
69 specBuilder *scheduler.SpecBuilder,
70 > ) { generator.go
71 > futureTimes, err := g.computeFutureActionTimes(ctx, specBuilder)
72 > if err != nil {
73 g.getOrCreateEventLog(ctx).LogEvent(ctx,
74 fmt.Sprintf("failed to update future action times: %v", err.Error()))
76 return
77 }
78 > g.FutureActionTimes = futureTimes generator.go
79 }
80
84 ctx chasm.Context,
85 specBuilder *scheduler.SpecBuilder,
86 > ) ([]*timestamppb.Timestamp, error) { generator.go
87 > sched := g.Scheduler.Get(ctx)
88 > spec, err := sched.getCompiledSpec(specBuilder)
89 > if err != nil {
90 return nil, err
91 }
92
93 > count := recentActionCount generator.go
94 > if sched.Schedule.State.LimitedActions {
95 count = min(int(sched.Schedule.State.RemainingActions), recentActionCount)
96 }
97
98 > futureTimes := make([]*timestamppb.Timestamp, 0, count) generator.go
99 > // Start from max(now, updateTime) to ensure we skip times before the last update.
100 > t := ctx.Now(g)
101 > if updateTime := sched.Info.GetUpdateTime().AsTime(); updateTime.After(t) {
102 t = updateTime
103 }
104 > for len(futureTimes) < count { generator.go
105 > res, err := spec.GetNextTime(sched.jitterSeed(), t)
106 > if err != nil || res.Next.IsZero() {
107 // Over-excluded spec (limit) or end of schedule: return a partial list.
108 break
109 }
110 > t = res.Next generator.go
111 > futureTimes = append(futureTimes, timestamppb.New(t))
112 }
113
114 > return futureTimes, nil generator.go
115 }
go.temporal.io/server/chasm/lib/scheduler/invoker.go 15 introduced LOC · 5 ranges

Open complete file

39 }
40
41 > func newInvokerWithState(ctx chasm.MutableContext, state *schedulerpb.InvokerState) *Invoker { invoker.go
42 > i := &Invoker{
43 > InvokerState: state,
44 > EventLog: chasm.NewComponentField(ctx, NewEventLog(ctx)),
45 > }
46 > return i
47 > }
48
49 // EnqueueBufferedStarts adds new BufferedStarts to the invocation queue,
378 // runningWorkflowExecutions returns the list of workflow executions that
379 // have been started but not yet completed.
380 > func (i *Invoker) runningWorkflowExecutions() []*commonpb.WorkflowExecution { invoker.go
381 > var running []*commonpb.WorkflowExecution
382 > for _, start := range i.GetBufferedStarts() {
383 if start.GetRunId() != "" && start.GetCompleted() == nil {
384 running = append(running, &commonpb.WorkflowExecution{
388 }
389 }
390 > return running invoker.go
391 }
392
394 // This includes both running workflows (with status RUNNING) and completed
395 // workflows (with their final status).
396 > func (i *Invoker) recentActions() []*schedulepb.ScheduleActionResult { invoker.go
397 > var results []*schedulepb.ScheduleActionResult
398 > for _, start := range i.GetBufferedStarts() {
399 // Only include workflows that have been started (have a RunId).
400 if start.GetRunId() == "" {
415 })
416 }
417 > return results invoker.go
418 }
419
go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1/tasks.pb.go 8 introduced LOC · 1 range

Open complete file

131 func (*GeneratorTask) ProtoMessage() {}
132
133 > func (x *GeneratorTask) ProtoReflect() protoreflect.Message { tasks.pb.go
134 > mi := &file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_msgTypes[2]
135 > if x != nil {
136 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
137 > if ms.LoadMessageInfo() == nil {
138 > ms.StoreMessageInfo(mi)
139 > }
140 > return ms
141 }
142 return mi.MessageOf(x)
go.temporal.io/server/chasm/lib/scheduler/util.go 7 introduced LOC · 3 ranges

Open complete file

70 lastProcessedTime *timestamppb.Timestamp,
71 scheduledAt time.Time,
72 > ) (bool, error) { util.go
73 > // Immediate tasks are always valid - they execute inline during the transaction.
74 > if scheduledAt.IsZero() {
75 > return true, nil
76 > }
77 // If lastProcessedTime is not set, all scheduled tasks are valid.
78 > if lastProcessedTime == nil || (lastProcessedTime.GetSeconds() == 0 && lastProcessedTime.GetNanos() == 0) { util.go
79 return true, nil
80 }
81 // Scheduled tasks are valid if their time is after the high water mark.
82 > return scheduledAt.After(lastProcessedTime.AsTime()), nil util.go
83 }
84
go.temporal.io/server/chasm/search_attribute.go 7 introduced LOC · 1 range

Open complete file

314
315 // Value sets the integer value of the search attribute.
316 > func (s SearchAttributeInt) Value(value int64) SearchAttributeKeyValue { search_attribute.go
317 > return SearchAttributeKeyValue{
318 > Alias: s.alias,
319 > Field: s.field,
320 > Value: VisibilityValueInt64(value),
321 > }
322 > }
323
324 func (s SearchAttributeInt) typeMarker(_ int64) {}
go.temporal.io/server/chasm/lib/scheduler/eventlog.go 5 introduced LOC · 1 range

Open complete file

37 }
38
39 > func (g *Generator) getOrCreateEventLog(ctx chasm.MutableContext) *EventLog { eventlog.go
40 > eventLog, ok := g.EventLog.TryGet(ctx)
41 > if ok {
42 > return eventLog
43 > }
44 eventLog = NewEventLog(ctx)
45 g.EventLog = chasm.NewComponentField(ctx, eventLog)
go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1/message.pb.go 5 introduced LOC · 1 range

Open complete file

487 mi := &file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_msgTypes[5]
488 if x != nil {
489 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) message.pb.go
490 > if ms.LoadMessageInfo() == nil {
491 > ms.StoreMessageInfo(mi)
492 > }
493 > return ms
494 }
495 return mi.MessageOf(x)

Direct hierarchy

Parents

Children