workflow.go ×10

Frontier kind: Code frontier

unlabeled · c_697c9bae9909

43 tests · 2712 LOC · 120 files · introduces 0 tests · 71 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
13 ranges71 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
447 ranges2712 lines · 120 files · Browse complete extent
All tests (intent)
43 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: 71 introduced LOC across 13 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/scheduler/workflow.go 53 introduced LOC · 10 ranges

Open complete file

436 s.Info.InvalidScheduleError = err.Error()
437 s.cspec = nil
438 > } else { workflow.go
439 > s.Info.InvalidScheduleError = ""
440 > s.cspec = cspec
441 > }
442 }
443
545 // refilled, try to refill it starting from `cacheBase` instead of `after`. This avoids having
546 // the cache range jump back and forth when generating a sequence of times.
547 > func (s *scheduler) getNextTimeV2(cacheBase, after time.Time) GetNextTimeResult { workflow.go
548 > // cacheBase must be before after
549 > cacheBase = util.MinTime(cacheBase, after)
550 >
551 > // Asking for a time before the cache, need to refill.
552 > // Also if version changed (so we can fix a bug immediately).
553 > if s.nextTimeCacheV2 == nil ||
554 > after.Before(s.nextTimeCacheV2.StartTime.AsTime()) ||
555 > SchedulerWorkflowVersion(s.nextTimeCacheV2.Version) != s.tweakables.Version {
556 > s.fillNextTimeCacheV2(cacheBase)
557 > }
558
559 // We may need up to three tries: the first is in the cache as it exists now,
560 // the second is refilled from cacheBase, and the third is if cacheBase was set
561 // too far in the past, we ignore it and fill the cache from after.
562 > for try := 1; try <= 3; try++ { workflow.go
563 > if res, ok := searchCache(s.nextTimeCacheV2, after); ok {
564 > return res
565 > }
566 // Otherwise refill from base
567 s.fillNextTimeCacheV2(cacheBase)
576 }
577
578 > func searchCache(cache *schedulespb.NextTimeCache, after time.Time) (GetNextTimeResult, bool) { workflow.go
579 > // The cache covers a contiguous time range so we can do a linear search in it.
580 > start := cache.StartTime.AsTime()
581 > afterOffset := int64(after.Sub(start))
582 > for i, nextOffset := range cache.NextTimes {
583 if nextOffset > afterOffset {
584 next := start.Add(time.Duration(nextOffset))
597 }
598
599 > func (s *scheduler) fillNextTimeCacheV2(start time.Time) { workflow.go
600 > // Clear value so we can Get into it
601 > s.nextTimeCacheV2 = nil
602 > // Run this logic in a SideEffect so that we can fix bugs there without breaking
603 > // existing schedule workflows.
604 > val := workflow.SideEffect(s.ctx, func(ctx workflow.Context) any {
605 cache := &schedulespb.NextTimeCache{
606 Version: int64(s.tweakables.Version),
641 // but it might not. To be sure, check StartTime also (the field names differ between json
642 // and proto so json.Unmarshal will never fill in StartTime).
643 > if val.Get(&s.nextTimeCacheV2) == nil && s.nextTimeCacheV2.GetStartTime() != nil { workflow.go
644 > return
645 > }
646 // Try as json value
647 var jsonVal jsonNextTimeCacheV2
662 }
663
664 > func (s *scheduler) getNextTime(after time.Time) GetNextTimeResult { workflow.go
665 > // Implementation using a cache to save markers + computation.
666 > if s.hasMinVersion(NewCacheAndJitter) {
667 > return s.getNextTimeV2(after, after)
668 > } else if s.hasMinVersion(BatchAndCacheTimeQueries) {
669 return s.getNextTimeV1(after)
670 }
691 }
692
693 > catchupWindow := s.getCatchupWindow() workflow.go
694 >
695 > // A previous version would record a marker for each time which could make a workflow
696 > // fail. With the new version, the entire time range is skipped if the workflow is paused
697 > // or we are not going to take an action now
698 > if s.hasMinVersion(BatchAndCacheTimeQueries) {
699 > // Peek at paused/remaining actions state and don't bother if we're not going to
700 > // take an action now. (Don't count as missed catchup window either.)
701 > // Skip over entire time range if paused or no actions can be taken
702 > if !s.canTakeScheduledAction(manual, false) {
703 // use end as last action time so that we don't reprocess time spent paused
704 return s.getNextTime(end).Next, end
1337 } else if cw.AsDuration() < s.tweakables.MinCatchupWindow {
1338 return s.tweakables.MinCatchupWindow
1339 > } else { workflow.go
1340 > return cw.AsDuration()
1341 > }
1342 }
1343
1577 }
1578
1579 > func (s *scheduler) jitterSeed() string { workflow.go
1580 > if s.hasMinVersion(NewCacheAndJitter) {
1581 return fmt.Sprintf("%s-%s", s.State.NamespaceId, s.State.ScheduleId)
1582 }
go.temporal.io/server/api/schedule/v1/message.pb.go 18 introduced LOC · 3 ranges

Open complete file

986 }
987
988 > func (x *NextTimeCache) Reset() { message.pb.go
989 > *x = NextTimeCache{}
990 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[12]
991 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
992 > ms.StoreMessageInfo(mi)
993 > }
994
995 func (x *NextTimeCache) String() string {
999 func (*NextTimeCache) ProtoMessage() {}
1000
1001 > func (x *NextTimeCache) ProtoReflect() protoreflect.Message { message.pb.go
1002 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[12]
1003 > if x != nil {
1004 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1005 > if ms.LoadMessageInfo() == nil {
1006 > ms.StoreMessageInfo(mi)
1007 > }
1008 > return ms
1009 }
1010 return mi.MessageOf(x)
1023 }
1024
1025 > func (x *NextTimeCache) GetStartTime() *timestamppb.Timestamp { message.pb.go
1026 > if x != nil {
1027 > return x.StartTime
1028 > }
1029 return nil
1030 }