Vihren Atlas

workflow.go ×24

Code frontier

unlabeled · c_355240e3976f

37 tests · 3050 LOC · 122 files · introduces 0 tests · 108 LOC · 1 file

Introduces — evidence that enters the hierarchy at this concept

Code
24 ranges108 lines · 1 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
530 ranges3050 lines · 122 files · Browse complete extent
All tests (intent)
37 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: 108 introduced LOC across 24 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/worker/scheduler/workflow.go 108 introduced LOC · 24 ranges

Open complete file

180 )
181
182 > func defaultLocalActivityOptions() workflow.LocalActivityOptions { workflow.go
183 > return workflow.LocalActivityOptions{
184 > // This applies to watch, cancel, and terminate. Start workflow overrides this.
185 > ScheduleToCloseTimeout: 1 * time.Hour,
186 > // Each local activity is one or a few local RPCs.
187 > // Currently this is applied manually, see https://github.com/temporalio/sdk-go/issues/1066
188 > StartToCloseTimeout: 5 * time.Second,
189 > RetryPolicy: &temporal.RetryPolicy{
190 > InitialInterval: 1 * time.Second,
191 > MaximumInterval: 60 * time.Second,
192 > },
193 > }
194 > }
195
196 var (
1383 // long-poll watcher running, because we would have gotten woken up already.
1384 if s.State.NeedRefresh {
1385 > s.refreshWorkflows(slices.Clone(s.Info.RunningWorkflows)) workflow.go
1386 > s.State.NeedRefresh = false
1387 > }
1388
1389 // Make sure we have something to start. If not, we can clear the buffer.
1394 }
1395
1396 > isRunning := len(s.Info.RunningWorkflows) > 0 workflow.go
1397 > tryAgain := false
1398 > action := ProcessBuffer(s.State.BufferedStarts, isRunning, s.resolveOverlapPolicy)
1399 >
1400 > s.State.BufferedStarts = action.NewBuffer
1401 > s.Info.OverlapSkipped += action.OverlapSkipped
1402 > for overlapPolicy, count := range action.OverlapSkippedByPolicy {
1403 s.metrics.WithTags(map[string]string{
1404 metrics.ScheduleOverlapPolicyTag: overlapPolicy.String(),
1407
1408 // Try starting whatever we're supposed to start now
1409 > allStarts := action.OverlappingStarts workflow.go
1410 > if action.NonOverlappingStart != nil {
1411 allStarts = append(allStarts, action.NonOverlappingStart)
1412 }
1413 > for _, start := range allStarts { workflow.go
1414 > if !s.canTakeScheduledAction(start.Manual, true) {
1415 // try again to drain the buffer if paused or out of actions
1416 tryAgain = true
1417 continue
1418 }
1419 > result, err := s.startWorkflow(start, req) workflow.go
1420 > metricsWithTag := s.metrics.WithTags(map[string]string{
1421 > metrics.ScheduleActionTypeTag: metrics.ScheduleActionStartWorkflow,
1422 > })
1423 > if err != nil {
1424 s.logger.Error("Failed to start workflow", "error", err)
1425 if !isUserScheduleError(err) {
1435
1436 // Terminate or cancel if required (terminate overrides cancel if both are present)
1437 > if action.NeedTerminate { workflow.go
1438 for _, ex := range s.Info.RunningWorkflows {
1439 s.terminateWorkflow(ex)
1440 }
1441 > } else if action.NeedCancel { workflow.go
1442 for _, ex := range s.Info.RunningWorkflows {
1443 s.cancelWorkflow(ex)
1449 // one of them with an activity. We only need one watcher at a time, though: after that one
1450 // returns, we'll end up back here and start the next one.
1451 > if len(s.State.BufferedStarts) > 0 && s.watchingFuture == nil { workflow.go
1452 if len(s.Info.RunningWorkflows) > 0 {
1453 s.startLongPollWatcher(s.Info.RunningWorkflows[0])
1457 }
1458
1459 > return tryAgain workflow.go
1460 }
1461
1472 start *schedulespb.BufferedStart,
1473 newWorkflow *workflowpb.NewWorkflowExecutionInfo,
1474 > ) (*schedulepb.ScheduleActionResult, error) { workflow.go
1475 > nominalTimeSec := start.NominalTime.AsTime().UTC().Truncate(time.Second)
1476 > workflowID := newWorkflow.WorkflowId
1477 > if start.OverlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL || s.tweakables.AlwaysAppendTimestamp {
1478 > // must match AppendedTimestampForValidation
1479 > workflowID += "-" + nominalTimeSec.Format(time.RFC3339)
1480 > }
1481
1482 // Set scheduleToCloseTimeout based on catchup window, which is the latest time that it's
1483 // acceptable to start this workflow. For manual starts (trigger immediately or backfill),
1484 // catch up window doesn't apply, so just use 60s.
1485 > options := defaultLocalActivityOptions() workflow.go
1486 > if start.Manual {
1487 options.ScheduleToCloseTimeout = 60 * time.Second
1488 > } else { workflow.go
1489 deadline := start.ActualTime.AsTime().Add(s.getCatchupWindow())
1490 options.ScheduleToCloseTimeout = deadline.Sub(s.now())
1495 }
1496 }
1497 > ctx := workflow.WithLocalActivityOptions(s.ctx, options) workflow.go
1498 >
1499 > lastCompletionResult, continuedFailure := s.State.LastCompletionResult, s.State.ContinuedFailure
1500 > if start.OverlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL && s.hasMinVersion(DontTrackOverlapping) {
1501 // ALLOW_ALL runs don't participate in lastCompletionResult/continuedFailure at all
1502 lastCompletionResult = nil
1506 // Reject duplicates as part of WFID reuse policy when possible, as a measure
1507 // against WFT timeouts/failures that lead to non-determinism.
1508 > reusePolicy := enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE workflow.go
1509 > if start.Manual {
1510 reusePolicy = enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE
1511 }
1512
1513 > req := &schedulespb.StartWorkflowRequest{ workflow.go
1514 > Request: &workflowservice.StartWorkflowExecutionRequest{
1515 > WorkflowId: workflowID,
1516 > WorkflowType: newWorkflow.WorkflowType,
1517 > TaskQueue: newWorkflow.TaskQueue,
1518 > Input: newWorkflow.Input,
1519 > WorkflowExecutionTimeout: newWorkflow.WorkflowExecutionTimeout,
1520 > WorkflowRunTimeout: newWorkflow.WorkflowRunTimeout,
1521 > WorkflowTaskTimeout: newWorkflow.WorkflowTaskTimeout,
1522 > Identity: s.identity(),
1523 > RequestId: s.newUUIDString(),
1524 > WorkflowIdReusePolicy: reusePolicy,
1525 > RetryPolicy: newWorkflow.RetryPolicy,
1526 > Memo: newWorkflow.Memo,
1527 > SearchAttributes: s.addSearchAttributes(newWorkflow.SearchAttributes, nominalTimeSec),
1528 > Header: newWorkflow.Header,
1529 > LastCompletionResult: lastCompletionResult,
1530 > ContinuedFailure: continuedFailure,
1531 > UserMetadata: newWorkflow.UserMetadata,
1532 > Priority: newWorkflow.Priority,
1533 > },
1534 > }
1535 > for {
1536 > var res schedulespb.StartWorkflowResponse
1537 > err := workflow.ExecuteLocalActivity(ctx, s.a.StartWorkflow, req).Get(s.ctx, &res)
1538 > var appErr *temporal.ApplicationError
1539 > var details rateLimitedDetails
1540 > if errors.As(err, &appErr) && appErr.Type() == rateLimitedErrorType && appErr.Details(&details) == nil {
1541 s.metrics.Counter(metrics.ScheduleRateLimited.Name()).Inc(1)
1542 workflow.Sleep(s.ctx, details.Delay)
1544 continue
1545 }
1546 > if err != nil { workflow.go
1547 return nil, err
1548 }
1573 }
1574
1575 > func (s *scheduler) identity() string { workflow.go
1576 > return fmt.Sprintf("temporal-scheduler-%s-%s", s.State.Namespace, s.State.ScheduleId)
1577 > }
1578
1579 func (s *scheduler) jitterSeed() string {
1587 attributes *commonpb.SearchAttributes,
1588 nominal time.Time,
1589 > ) *commonpb.SearchAttributes { workflow.go
1590 > fields := util.CloneMapNonNil(attributes.GetIndexedFields())
1591 > if p, err := payload.Encode(nominal); err == nil {
1592 > fields[sadefs.TemporalScheduledStartTime] = p
1593 > }
1594 > if p, err := payload.Encode(s.State.ScheduleId); err == nil {
1595 > fields[sadefs.TemporalScheduledById] = p
1596 > }
1597 > return &commonpb.SearchAttributes{
1598 > IndexedFields: fields,
1599 > }
1600 }
1601
1602 > func (s *scheduler) refreshWorkflows(executions []*commonpb.WorkflowExecution) { workflow.go
1603 > ctx := workflow.WithLocalActivityOptions(s.ctx, defaultLocalActivityOptions())
1604 > futures := make([]workflow.Future, len(executions))
1605 > for i, ex := range executions {
1606 req := &schedulespb.WatchWorkflowRequest{
1607 // Note: do not send runid here so that we always get the latest one
1612 futures[i] = workflow.ExecuteLocalActivity(ctx, s.a.WatchWorkflow, req)
1613 }
1614 > for i, ex := range executions { workflow.go
1615 s.processWatcherResult(ex.WorkflowId, futures[i], false)
1616 }
1705 }
1706
1707 > func (s *scheduler) newUUIDString() string { workflow.go
1708 > if len(s.uuidBatch) == 0 {
1709 > panicIfErr(workflow.SideEffect(s.ctx, func(ctx workflow.Context) any {
1710 out := make([]string, 10)
1711 for i := range out {
1715 }).Get(&s.uuidBatch))
1716 }
1717 > next := s.uuidBatch[0] workflow.go
1718 > s.uuidBatch = s.uuidBatch[1:]
1719 > return next
1720 }
1721
1724 }
1725
1726 > func panicIfErr(err error) { workflow.go
1727 > if err != nil {
1728 panic(err)
1729 }

Direct hierarchy

Parents

Children