Vihren Atlas

workflow.go ×33

Code frontier

unlabeled · c_37d125cba9e6

44 tests · 2503 LOC · 119 files · introduces 0 tests · 141 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
37 ranges141 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
390 ranges2503 lines · 119 files · Browse complete extent
All tests (intent)
44 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.

2 files ranked by introduced lines: 141 introduced LOC across 37 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/scheduler/workflow.go 123 introduced LOC · 33 ranges

Open complete file

239 }
240
241 > func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.StartScheduleArgs, specBuilder *SpecBuilder, enableCHASMMigration func() bool, migrateWithRunningWorkflows func() bool) error { workflow.go
242 > scheduler := &scheduler{
243 > StartScheduleArgs: args,
244 > ctx: ctx,
245 > a: nil,
246 > logger: sdklog.With(workflow.GetLogger(ctx), "wf-namespace", args.State.Namespace, "schedule-id", args.State.ScheduleId),
247 > metrics: workflow.GetMetricsHandler(ctx).WithTags(map[string]string{
248 > "namespace": args.State.Namespace,
249 > metrics.ScheduleBackendTag: metrics.ScheduleBackendLegacy,
250 > }),
251 > specBuilder: specBuilder,
252 > enableCHASMMigration: enableCHASMMigration,
253 > migrateWithRunningWorkflows: migrateWithRunningWorkflows,
254 > }
255 > return scheduler.run()
256 > }
257
258 > func (s *scheduler) run() error { workflow.go
259 > s.updateTweakables()
260 > s.ensureFields()
261 > s.compileSpec()
262 >
263 > if err := workflow.SetQueryHandler(s.ctx, QueryNameDescribe, s.handleDescribeQuery); err != nil {
264 return err
265 }
266 > if err := workflow.SetQueryHandler(s.ctx, QueryNameListMatchingTimes, s.handleListMatchingTimesQuery); err != nil { workflow.go
267 return err
268 }
269
270 > if s.State.LastProcessedTime == nil { workflow.go
271 // log these as json since it's more readable than the Go representation
272 specJson, _ := protojson.Marshal(s.Schedule.Spec)
283 // A schedule may be created with an initial Patch, e.g. start one immediately. Put that in
284 // the state so it takes effect below.
285 > s.pendingPatch = s.InitialPatch workflow.go
286 > s.InitialPatch = nil
287 >
288 > iters := s.tweakables.IterationsBeforeContinueAsNew
289 > for {
290 > info := workflow.GetInfo(s.ctx)
291 > suggestContinueAsNew := info.GetCurrentHistoryLength() >= impossibleHistorySize
292 > if s.tweakables.IterationsBeforeContinueAsNew > 0 {
293 // forceCAN must be checked here too, not just in the else branch,
294 // so that the force-continue-as-new signal is honored regardless
296 suggestContinueAsNew = suggestContinueAsNew || iters <= 0 || s.forceCAN
297 iters--
298 > } else { workflow.go
299 suggestContinueAsNew = suggestContinueAsNew || info.GetContinueAsNewSuggested() || s.forceCAN
300 }
301 > if suggestContinueAsNew && s.pendingUpdate == nil && s.pendingPatch == nil { workflow.go
302 break
303 }
304
305 > t1 := timestamp.TimeValue(s.State.LastProcessedTime) workflow.go
306 > t2 := s.now()
307 > if t2.Before(t1) {
308 // Time went backwards. Currently this can only happen across a continue-as-new boundary.
309 s.logger.Warn("Time went backwards", "from", t1, "to", t2)
310 t2 = t1
311 }
312 > nextWakeup, lastAction := s.processTimeRange( workflow.go
313 > t1, t2,
314 > // resolve this to the schedule's policy as late as possible
315 > enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED,
316 > false,
317 > nil,
318 > )
319 > if s.hasMinVersion(UseLastAction) {
320 > s.State.LastProcessedTime = timestamppb.New(lastAction)
321 > } else {
322 s.State.LastProcessedTime = timestamppb.New(t2)
323 }
324 // handle signals after processing time range that just elapsed
325 > scheduleChanged := s.processSignals() workflow.go
326 > if scheduleChanged {
327 // need to calculate sleep again
328 nextWakeup, _ = s.processTimeRange(
335 }
336
337 > if !s.State.PendingMigration && s.tweakables.EnableCHASMMigration && workflow.go
338 > (s.tweakables.MigrateWithRunningWorkflows || len(s.Info.RunningWorkflows) == 0) {
339 s.State.PendingMigration = true
340 }
341 > if s.State.PendingMigration { workflow.go
342 err := s.executeMigration()
343 if err == nil {
395 }
396
397 > func (s *scheduler) ensureFields() { workflow.go
398 > if s.Schedule == nil {
399 s.Schedule = &schedulepb.Schedule{}
400 }
401 > if s.Schedule.Spec == nil { workflow.go
402 s.Schedule.Spec = &schedulepb.ScheduleSpec{}
403 }
404 > if s.Schedule.Action == nil { workflow.go
405 s.Schedule.Action = &schedulepb.ScheduleAction{}
406 }
407 > if s.Schedule.Policies == nil { workflow.go
408 s.Schedule.Policies = &schedulepb.SchedulePolicies{}
409 }
410
411 // set defaults eagerly so they show up in describe output
412 > s.Schedule.Policies.OverlapPolicy = s.resolveOverlapPolicy(s.Schedule.Policies.OverlapPolicy) workflow.go
413 > s.Schedule.Policies.CatchupWindow = durationpb.New(s.getCatchupWindow())
414 >
415 > if s.Schedule.State == nil {
416 s.Schedule.State = &schedulepb.ScheduleState{}
417 }
418 > if s.Info == nil { workflow.go
419 > s.Info = &schedulepb.ScheduleInfo{}
420 > }
421 > if s.State == nil {
422 s.State = &schedulespb.InternalState{}
423 }
424 }
425
426 > func (s *scheduler) compileSpec() { workflow.go
427 > // if spec changes invalidate current cache
428 > s.nextTimeCacheV1 = nil
429 > s.nextTimeCacheV2 = nil
430 >
431 > cspec, err := s.specBuilder.NewCompiledSpec(s.Schedule.Spec)
432 > if err != nil {
433 if s.logger != nil {
434 s.logger.Error("Invalid schedule", "error", err)
442 }
443
444 > func (s *scheduler) now() time.Time { workflow.go
445 > // Notes:
446 > // 1. The time returned here is actually the timestamp of the WorkflowTaskStarted
447 > // event, which is generated in history, not any time on the worker itself.
448 > // 2. There will be some delay between when history stamps the time on the
449 > // WorkflowTaskStarted event and when this code runs, as the event+task goes through
450 > // matching and frontend. But it should be well under a second, which is our minimum
451 > // granularity anyway.
452 > // 3. It's actually the maximum of all of those events: the go sdk enforces that
453 > // workflow time is monotonic. So if the clock on a history node is wrong and workflow
454 > // time is ahead of real time for a while, and then goes back, scheduled jobs will run
455 > // ahead of time, and then nothing will happen until real time catches up to where it
456 > // was temporarily ahead. Currently the only way to "recover" from this situation is
457 > // to recreate the schedule/scheduler workflow.
458 > // 4. Actually there is one way time can appear to go backwards from the point of view
459 > // of this workflow: across a continue-as-new, since monotonicity isn't preserved
460 > // there (as far as I know). We'll treat that the same since we keep track of the last
461 > // processed schedule time.
462 > return workflow.Now(s.ctx)
463 > }
464
465 func (s *scheduler) processPatch(patch *schedulepb.SchedulePatch) {
684 manual bool,
685 limit *int,
686 > ) (time.Time, time.Time) { workflow.go
687 > s.logger.Debug("processTimeRange", "start", start, "end", end, "overlap-policy", overlapPolicy, "manual", manual)
688 >
689 > if s.cspec == nil {
690 return time.Time{}, end
691 }
749 }
750
751 > func (s *scheduler) canTakeScheduledAction(manual, decrement bool) bool { workflow.go
752 > // If manual (trigger immediately or backfill), always allow
753 > if manual {
754 return true
755 }
756 // If paused, don't do anything
757 > if s.Schedule.State.Paused { workflow.go
758 return false
759 }
1086 }
1087
1088 > func (s *scheduler) processSignals() bool { workflow.go
1089 > scheduleChanged := false
1090 > if s.pendingPatch != nil {
1091 s.processPatch(s.pendingPatch)
1092 s.pendingPatch = nil
1093 }
1094 > if s.pendingUpdate != nil { workflow.go
1095 s.processUpdate(s.pendingUpdate)
1096 s.pendingUpdate = nil
1097 scheduleChanged = true
1098 }
1099 > return scheduleChanged workflow.go
1100 }
1101
1316 }
1317
1318 > func (s *scheduler) updateTweakables() { workflow.go
1319 > // Use MutableSideEffect so that we can change the defaults without breaking determinism.
1320 > get := func(ctx workflow.Context) any {
1321 p := CurrentTweakablePolicies
1322 // Re-evaluates migration dynamic config each iteration.
1325 return p
1326 }
1327 > eq := func(a, b any) bool { return a.(TweakablePolicies) == b.(TweakablePolicies) } workflow.go
1328 > if err := workflow.MutableSideEffect(s.ctx, "tweakables", get, eq).Get(&s.tweakables); err != nil {
1329 panic("can't decode TweakablePolicies:" + err.Error())
1330 }
1331 }
1332
1333 > func (s *scheduler) getCatchupWindow() time.Duration { workflow.go
1334 > cw := s.Schedule.Policies.CatchupWindow
1335 > if cw == nil {
1336 return s.tweakables.DefaultCatchupWindow
1337 > } else if cw.AsDuration() < s.tweakables.MinCatchupWindow { workflow.go
1338 return s.tweakables.MinCatchupWindow
1339 } else {
1342 }
1343
1344 > func (s *scheduler) resolveOverlapPolicy(overlapPolicy enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy { workflow.go
1345 > if overlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {
1346 > overlapPolicy = s.Schedule.Policies.OverlapPolicy
1347 > }
1348 > if overlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {
1349 overlapPolicy = enumspb.SCHEDULE_OVERLAP_POLICY_SKIP
1350 }
1351 > return overlapPolicy workflow.go
1352 }
1353
1720 }
1721
1722 > func (s *scheduler) hasMinVersion(version SchedulerWorkflowVersion) bool { workflow.go
1723 > return s.tweakables.Version >= version
1724 > }
1725
1726 func panicIfErr(err error) {
go.temporal.io/server/api/schedule/v1/message.pb.go 18 introduced LOC · 4 ranges

Open complete file

284 func (*InternalState) ProtoMessage() {}
285
286 > func (x *InternalState) ProtoReflect() protoreflect.Message { message.pb.go
287 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[2]
288 > if x != nil {
289 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
290 if ms.LoadMessageInfo() == nil {
293 return ms
294 }
295 > return mi.MessageOf(x) message.pb.go
296 }
297
388 }
389
390 > func (x *StartScheduleArgs) Reset() { message.pb.go
391 > *x = StartScheduleArgs{}
392 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[3]
393 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
394 > ms.StoreMessageInfo(mi)
395 > }
396
397 func (x *StartScheduleArgs) String() string {
401 func (*StartScheduleArgs) ProtoMessage() {}
402
403 > func (x *StartScheduleArgs) ProtoReflect() protoreflect.Message { message.pb.go
404 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[3]
405 > if x != nil {
406 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
407 > if ms.LoadMessageInfo() == nil {
408 > ms.StoreMessageInfo(mi)
409 > }
410 > return ms
411 }
412 return mi.MessageOf(x)

Direct hierarchy

Parents

Children