Vihren Atlas

workflow.go ×8

Code frontier

unlabeled · c_03ca63c5c022

5 tests · 3330 LOC · 122 files · introduces 0 tests · 61 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
10 ranges61 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
612 ranges3330 lines · 122 files · Browse complete extent
All tests (intent)
5 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: 61 introduced LOC across 10 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/scheduler/workflow.go 47 introduced LOC · 8 ranges

Open complete file

325 scheduleChanged := s.processSignals()
326 if scheduleChanged {
327 > // need to calculate sleep again workflow.go
328 > nextWakeup, _ = s.processTimeRange(
329 > s.State.LastProcessedTime.AsTime(),
330 > t2,
331 > enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED,
332 > false,
333 > nil,
334 > )
335 > }
336
337 if !s.State.PendingMigration && s.tweakables.EnableCHASMMigration &&
779 upCh := workflow.GetSignalChannel(s.ctx, SignalNameUpdate)
780 sel.AddReceive(upCh, func(ch workflow.ReceiveChannel, _ bool) {
781 > ch.Receive(s.ctx, &s.pendingUpdate) workflow.go
782 > })
783
784 reqCh := workflow.GetSignalChannel(s.ctx, SignalNamePatch)
993 }
994
995 > func (s *scheduler) processUpdate(req *schedulespb.FullUpdateRequest) { workflow.go
996 > if err := s.checkConflict(req.ConflictToken); err != nil {
997 s.logger.Warn("Update conflicted with concurrent change")
998 return
999 }
1000
1001 > s.logger.Debug("Schedule update") workflow.go
1002 >
1003 > s.Schedule.Spec = req.Schedule.GetSpec()
1004 > s.Schedule.Action = req.Schedule.GetAction()
1005 > s.Schedule.Policies = req.Schedule.GetPolicies()
1006 > s.Schedule.State = req.Schedule.GetState()
1007 > // don't touch Info
1008 >
1009 > s.ensureFields()
1010 > s.compileSpec()
1011 >
1012 > s.updateCustomSearchAttributes(req.SearchAttributes)
1013 >
1014 > // Record customer start workflow memo payload size on each update.
1015 > s.recordActionPayloadMetrics()
1016 >
1017 > if s.hasMinVersion(UpdateFromPrevious) && !s.hasMinVersion(UseLastAction) {
1018 // We need to start re-processing from the last event, so that we catch actions whose
1019 // nominal time is before now but actual time (with jitter) is after now. Logic in
1025 }
1026
1027 > s.Info.UpdateTime = timestamppb.New(s.now()) workflow.go
1028 > s.incSeqNo()
1029 }
1030
1093 }
1094 if s.pendingUpdate != nil {
1095 > s.processUpdate(s.pendingUpdate) workflow.go
1096 > s.pendingUpdate = nil
1097 > scheduleChanged = true
1098 > }
1099 return scheduleChanged
1100 }
1221 }
1222
1223 > func (s *scheduler) updateCustomSearchAttributes(searchAttributes *commonpb.SearchAttributes) { workflow.go
1224 > // We want to distinguish nil value from an empty object value.
1225 > // When it is nil, then it's a no-op. Otherwise, it will overwrite the search attributes.
1226 > // That is, if it is an empty map, it will unset all search attributes.
1227 > if searchAttributes == nil {
1228 > return
1229 > }
1230
1231 upsertMap := map[string]any{}
1309 }
1310
1311 > func (s *scheduler) checkConflict(token int64) error { workflow.go
1312 > if token == 0 || token == s.State.ConflictToken {
1313 > return nil
1314 > }
1315 return errUpdateConflict
1316 }
go.temporal.io/server/api/schedule/v1/message.pb.go 14 introduced LOC · 2 ranges

Open complete file

455 }
456
457 > func (x *FullUpdateRequest) Reset() { message.pb.go
458 > *x = FullUpdateRequest{}
459 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[4]
460 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
461 > ms.StoreMessageInfo(mi)
462 > }
463
464 func (x *FullUpdateRequest) String() string {
468 func (*FullUpdateRequest) ProtoMessage() {}
469
470 > func (x *FullUpdateRequest) ProtoReflect() protoreflect.Message { message.pb.go
471 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[4]
472 > if x != nil {
473 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
474 > if ms.LoadMessageInfo() == nil {
475 > ms.StoreMessageInfo(mi)
476 > }
477 > return ms
478 }
479 return mi.MessageOf(x)

Direct hierarchy

Parents

Children