Vihren Atlas

activity.go ×12

Joint frontier

unlabeled · c_35908e3a2504

1 test · 2715 LOC · 129 files · introduces 1 test · 72 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
13 ranges72 lines · 2 files
Tests
1 test

Contains — complete concept membership

All code (extent)
472 ranges2715 lines · 129 files · Browse complete extent
All tests (intent)
1 testBrowse 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.

1 test introduced at this concept.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

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

go.temporal.io/server/chasm/lib/activity/activity.go 71 introduced LOC · 12 ranges

Open complete file

702 }
703
704 > updateFields := map[string]struct{}{} activity.go
705 > if mask := frontendReq.GetUpdateMask(); mask != nil {
706 > updateFields = util.ParseFieldMask(mask)
707 > }
708
709 > _, hasStartDelayInMask := updateFields["startDelay"] activity.go
710 > if hasStartDelayInMask {
711 newDelay := frontendReq.GetActivityOptions().GetStartDelay()
712 if err := validateStartDelay(newDelay); err != nil {
727 }
728
729 > attempt := a.LastAttempt.Get(ctx) activity.go
730 >
731 > if frontendReq.GetRestoreOriginal() {
732 ogOptions := a.GetOriginalOptions()
733 a.TaskQueue = common.CloneProto(ogOptions.GetTaskQueue())
743 a.StartDelay = common.CloneProto(ogOptions.GetStartDelay())
744 }
745 > } else { activity.go
746 > if err := a.mergeActivityOptions(frontendReq); err != nil {
747 return nil, err
748 }
751 // Recalculate policy-derived retry intervals based on the (possibly updated) retry policy.
752 // Worker-provided NextRetryDelay values are preserved for their already-scheduled retry.
753 > if a.shouldRecalculateCurrentRetryInterval(attempt, frontendReq.GetRestoreOriginal(), updateFields) { activity.go
754 newInterval := backoff.CalculateExponentialRetryInterval(a.RetryPolicy, attempt.GetCount()-1)
755 attempt.CurrentRetryInterval = durationpb.New(newInterval)
757
758 // Recreate the ScheduleToClose task at the (possibly updated) deadline.
759 > a.reissueScheduleToClose(ctx) activity.go
760 >
761 > attempt.Stamp++
762 >
763 > a.reissueRunningAttemptTimers(ctx, attempt)
764 > if a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED {
765 a.reissueDispatchAndScheduleToStart(ctx, attempt)
766 }
767
768 > metricsHandler, err := a.enrichMetricsHandler(ctx, metrics.ActivityUpdateOptionsScope) activity.go
769 > if err != nil {
770 return nil, err
771 }
772 > a.emitOnUpdateOptionsMetrics(metricsHandler) activity.go
773 >
774 > return &activitypb.UpdateActivityExecutionOptionsResponse{
775 > FrontendResponse: &workflowservice.UpdateActivityExecutionOptionsResponse{
776 > ActivityOptions: &apiactivitypb.ActivityOptions{
777 > TaskQueue: a.GetTaskQueue(),
778 > ScheduleToCloseTimeout: a.GetScheduleToCloseTimeout(),
779 > ScheduleToStartTimeout: a.GetScheduleToStartTimeout(),
780 > StartToCloseTimeout: a.GetStartToCloseTimeout(),
781 > HeartbeatTimeout: a.GetHeartbeatTimeout(),
782 > RetryPolicy: a.GetRetryPolicy(),
783 > Priority: a.GetPriority(),
784 > StartDelay: a.GetStartDelay(),
785 > },
786 > },
787 > }, nil
788 }
789
1458 // updated) timeouts. No-op unless the activity is in a status where a worker holds the task token
1459 // (STARTED / CANCEL_REQUESTED / PAUSE_REQUESTED / RESET_REQUESTED).
1460 > func (a *Activity) reissueRunningAttemptTimers(ctx chasm.MutableContext, attempt *activitypb.ActivityAttemptState) { activity.go
1461 > if !a.hasAttemptInProgress() {
1462 return
1463 }
1464 > if timeout := a.GetStartToCloseTimeout().AsDuration(); timeout > 0 { activity.go
1465 > deadline := attempt.GetStartedTime().AsTime().Add(timeout)
1466 > ctx.AddTask(
1467 > a,
1468 > chasm.TaskAttributes{ScheduledTime: deadline},
1469 > &activitypb.StartToCloseTimeoutTask{Stamp: attempt.GetStamp()},
1470 > )
1471 > }
1472 > if hbTimeout := a.GetHeartbeatTimeout().AsDuration(); hbTimeout > 0 {
1473 > // Next heartbeat fires at max(last recorded heartbeat, current attempt start) + heartbeat timeout.
1474 > lastHb, _ := a.LastHeartbeat.TryGet(ctx)
1475 > lastHbTime := util.MaxTime(
1476 > lastHb.GetRecordedTime().AsTime(),
1477 > attempt.GetStartedTime().AsTime(),
1478 > ).Add(hbTimeout)
1479 > ctx.AddTask(
1480 > a,
1481 > chasm.TaskAttributes{ScheduledTime: lastHbTime},
1482 > &activitypb.HeartbeatTimeoutTask{Stamp: attempt.GetStamp()},
1483 > )
1484 > }
1485 }
1486
1501 // reissueScheduleToClose bumps the ScheduleToCloseStamp and re-emits the ScheduleToClose timeout task
1502 // at the current deadline.
1503 > func (a *Activity) reissueScheduleToClose(ctx chasm.MutableContext) { activity.go
1504 > if deadline := a.scheduleToCloseDeadline(); !deadline.IsZero() {
1505 > a.ScheduleToCloseStamp++
1506 > ctx.AddTask(
1507 > a,
1508 > chasm.TaskAttributes{ScheduledTime: deadline},
1509 > &activitypb.ScheduleToCloseTimeoutTask{Stamp: a.GetScheduleToCloseStamp()},
1510 > )
1511 > }
1512 }
1513
2060 func (a *Activity) emitOnUpdateOptionsMetrics(
2061 handler metrics.Handler,
2062 > ) { activity.go
2063 > metrics.ActivityUpdateOptions.With(handler).Record(1)
2064 > }
2065
2066 func (a *Activity) emitOnUnpausedMetrics(
go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go 1 introduced LOC · 1 range

Open complete file

Direct hierarchy

Children