workflow.go ×15

Frontier kind: Code frontier

unlabeled · c_6a540a20c158

2 tests · 3805 LOC · 126 files · introduces 0 tests · 52 LOC · 1 file

Introduces — evidence that enters the hierarchy at this concept

Code
15 ranges52 lines · 1 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
765 ranges3805 lines · 126 files · Browse complete extent
All tests (intent)
2 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.

1 file ranked by introduced lines: 52 introduced LOC across 15 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/scheduler/workflow.go 52 introduced LOC · 15 ranges

Open complete file

320 s.State.LastProcessedTime = timestamppb.New(lastAction)
321 } else {
322 > s.State.LastProcessedTime = timestamppb.New(t2) workflow.go
323 > }
324 // handle signals after processing time range that just elapsed
325 scheduleChanged := s.processSignals()
494 }
495 s.State.OngoingBackfills = append(s.State.OngoingBackfills, common.CloneProto(bfr))
496 > } else { workflow.go
497 > // Old version: process whole backfill synchronously
498 > s.processTimeRange(
499 > timestamp.TimeValue(bfr.GetStartTime()),
500 > timestamp.TimeValue(bfr.GetEndTime()),
501 > bfr.GetOverlapPolicy(),
502 > true,
503 > nil,
504 > )
505 > }
506 }
507
518 }
519
520 > func (s *scheduler) getNextTimeV1(after time.Time) GetNextTimeResult { workflow.go
521 > // we populate the map sequentially, if after is not in the map, it means we either exhausted
522 > // all items, or we jumped through time (forward or backward), in either case, refresh the cache
523 > next, ok := s.nextTimeCacheV1[after]
524 > if ok {
525 > return next
526 > }
527 > s.nextTimeCacheV1 = nil
528 > // Run this logic in a SideEffect so that we can fix bugs there without breaking
529 > // existing schedule workflows.
530 > panicIfErr(workflow.SideEffect(s.ctx, func(ctx workflow.Context) any {
531 results := make(map[time.Time]GetNextTimeResult)
532 for t := after; !t.IsZero() && len(results) < nextTimeCacheV1Size; {
539 return results
540 }).Get(&s.nextTimeCacheV1))
541 > return s.nextTimeCacheV1[after] workflow.go
542 }
543
645 }
646 // Try as json value
647 > var jsonVal jsonNextTimeCacheV2 workflow.go
648 > if val.Get(&jsonVal) != nil || jsonVal.Start.IsZero() {
649 panic("could not decode next time cache as proto or json")
650 }
651 > s.nextTimeCacheV2 = &schedulespb.NextTimeCache{ workflow.go
652 > Version: int64(jsonVal.Version),
653 > StartTime: timestamppb.New(jsonVal.Start),
654 > NextTimes: make([]int64, len(jsonVal.Results)),
655 > NominalTimes: make([]int64, len(jsonVal.Results)),
656 > Completed: jsonVal.Completed,
657 > }
658 > for i, res := range jsonVal.Results {
659 > s.nextTimeCacheV2.NextTimes[i] = int64(res.Next.Sub(jsonVal.Start))
660 > s.nextTimeCacheV2.NominalTimes[i] = int64(res.Nominal.Sub(jsonVal.Start))
661 > }
662 }
663
667 return s.getNextTimeV2(after, after)
668 } else if s.hasMinVersion(BatchAndCacheTimeQueries) {
669 > return s.getNextTimeV1(after) workflow.go
670 > }
671 // Run this logic in a SideEffect so that we can fix bugs there without breaking
672 // existing schedule workflows.
673 > var next GetNextTimeResult workflow.go
674 > panicIfErr(workflow.SideEffect(s.ctx, func(ctx workflow.Context) any {
675 res, _ := s.cspec.GetNextTime(s.jitterSeed(), after)
676 return res
677 }).Get(&next))
678 > return next workflow.go
679 }
680
711 for next = s.getNextTime(start); !(next.Next.IsZero() || next.Next.After(end)); next = s.getNextTime(next.Next) {
712 if !s.hasMinVersion(BatchAndCacheTimeQueries) && !s.canTakeScheduledAction(manual, false) {
713 > continue workflow.go
714 }
715 if !manual && s.Info.UpdateTime.AsTime().After(next.Next) {
828 s.currentTimer = nil
829 })
830 > } else { workflow.go
831 > tmr := workflow.NewTimer(s.ctx, sleepTime)
832 > sel.AddFuture(tmr, func(_ workflow.Future) {})
833 }
834 }
841 sel.Select(s.ctx)
842 for sel.HasPending() {
843 > sel.Select(s.ctx) workflow.go
844 > }
845 }
846
1336 return s.tweakables.DefaultCatchupWindow
1337 } else if cw.AsDuration() < s.tweakables.MinCatchupWindow {
1338 > return s.tweakables.MinCatchupWindow workflow.go
1339 } else {
1340 return cw.AsDuration()
1453 s.startLongPollWatcher(s.Info.RunningWorkflows[0])
1454 } else {
1455 > s.logger.Error("have buffered workflows but none running") workflow.go
1456 > }
1457 }
1458
1581 return fmt.Sprintf("%s-%s", s.State.NamespaceId, s.State.ScheduleId)
1582 }
1583 > return "" workflow.go
1584 }
1585