diff --git a/service/history/workflow/activity.go b/service/history/workflow/activity.go index 12a0f984fd6..d29c6f9ad18 100644 --- a/service/history/workflow/activity.go +++ b/service/history/workflow/activity.go @@ -120,17 +120,15 @@ func GetPendingActivityInfo( p.Attempt = ai.Attempt if p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED { scheduledTime := ai.ScheduledTime.AsTime() - if now.Before(scheduledTime) { - // in this case activity is waiting for a retry + if now.Before(scheduledTime) && !ai.Paused { + // waiting for the retry to be dispatched to Matching p.NextAttemptScheduleTime = ai.ScheduledTime currentRetryDuration := p.NextAttemptScheduleTime.AsTime().Sub(p.LastAttemptCompleteTime.AsTime()) p.CurrentRetryInterval = durationpb.New(currentRetryDuration) } else { - // in this case activity is at least scheduled + // retry has been dispatched to Matching, or the activity is paused so no dispatch will occur p.NextAttemptScheduleTime = nil - // we rely on the fact that ExponentialBackoffAlgorithm is deterministic, and there's no random jitter - interval := backoff.ExponentialBackoffAlgorithm(ai.RetryInitialInterval, ai.RetryBackoffCoefficient, p.Attempt) - p.CurrentRetryInterval = durationpb.New(interval) + p.CurrentRetryInterval = nil } } } diff --git a/service/history/workflow/activity_test.go b/service/history/workflow/activity_test.go index d5d538b82ce..0959de85027 100644 --- a/service/history/workflow/activity_test.go +++ b/service/history/workflow/activity_test.go @@ -294,6 +294,36 @@ func (s *activitySuite) TestGetPendingActivityInfoHasRetryPolicy() { s.Equal(ai.RetryMaximumAttempts, pi.ActivityOptions.RetryPolicy.MaximumAttempts) } +func (s *activitySuite) TestGetPendingActivityInfoNextAttemptScheduleTimeAndCurrentRetryInterval() { + now := s.mockShard.GetTimeSource().Now().UTC() + activityType := commonpb.ActivityType{ + Name: "activityType", + } + ai := &persistencespb.ActivityInfo{ + StartedEventId: common.EmptyEventID, + LastAttemptCompleteTime: timestamppb.New(now), + HasRetryPolicy: true, + } + s.mockMutableState.EXPECT().GetActivityType(gomock.Any(), gomock.Any()).Return(&activityType, nil).Times(2) + + // Before dispatch to Matching: waiting for the retry, so we report when the next attempt is + // scheduled and the interval until then. + ai.ScheduledTime = timestamppb.New(now.Add(5 * time.Second)) + pi, err := GetPendingActivityInfo(context.Background(), s.mockShard, s.mockMutableState, ai) + s.NoError(err) + s.Equal(enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, pi.State) + s.Equal(ai.ScheduledTime, pi.NextAttemptScheduleTime) + s.Equal(durationpb.New(5*time.Second), pi.CurrentRetryInterval) + + // After dispatch to Matching: no next attempt schedule time or current retry interval. + ai.ScheduledTime = timestamppb.New(now.Add(-1 * time.Minute)) + pi, err = GetPendingActivityInfo(context.Background(), s.mockShard, s.mockMutableState, ai) + s.NoError(err) + s.Equal(enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, pi.State) + s.Nil(pi.NextAttemptScheduleTime) + s.Nil(pi.CurrentRetryInterval) +} + func (s *activitySuite) AddActivityInfo() *persistencespb.ActivityInfo { activityId := "activity-id" activityScheduledEvent := &historypb.HistoryEvent{ diff --git a/tests/activity_parity_test.go b/tests/activity_parity_test.go index 388e9b14cf3..822d39039ea 100644 --- a/tests/activity_parity_test.go +++ b/tests/activity_parity_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/require" commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" + failurepb "go.temporal.io/api/failure/v1" taskqueuepb "go.temporal.io/api/taskqueue/v1" "go.temporal.io/api/workflowservice/v1" sdkclient "go.temporal.io/sdk/client" @@ -17,6 +18,7 @@ import ( sdkworker "go.temporal.io/sdk/worker" "go.temporal.io/sdk/workflow" "go.temporal.io/server/common/retrypolicy" + "go.temporal.io/server/common/testing/await" "go.temporal.io/server/tests/testcore" "google.golang.org/protobuf/types/known/durationpb" ) @@ -59,6 +61,163 @@ type driver interface { // reproTimeout is the timeout under test, kept short so it fires within the test. const reproTimeout = 2 * time.Second +// current_retry_interval and next_attempt_schedule_time are reported while a retry is backing off +// (before it is dispatched to Matching), and for next_attempt_schedule_time also during start delay +// (SAA only). Once the attempt is dispatched, or while the activity is paused, both are nil. +func (s *standaloneActivityTestSuite) TestParityCurrentRetryInterval() { + env := s.newTestEnv() + t := s.T() + + both := func(t *testing.T, want activityInfoProjection, drive func(retryDriver, *testing.T) activityInfoProjection) { + t.Run("WorkflowActivity", func(t *testing.T) { + require.Equal(t, want, drive(&wfaDriver{s: s, env: env}, t)) + }) + t.Run("StandaloneActivity", func(t *testing.T) { + require.Equal(t, want, drive(&saaDriver{s: s, env: env}, t)) + }) + } + + // First attempt within its start delay (SAA only): the pending dispatch is in the future and is + // not a retry. + t.Run("StartDelayPending", func(t *testing.T) { + d := &saaDriver{s: s, env: env} + d.startWithStartDelay(t, startDelay) + info := d.describeActivity(t).GetInfo() + require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, info.GetRunState()) + require.Equal(t, info.GetExecutionTime().AsTime(), info.GetNextAttemptScheduleTime().AsTime(), + "during a start delay, NextAttemptScheduleTime is the pending dispatch time (schedule+delay)") + require.Nil(t, info.GetCurrentRetryInterval(), "the first attempt is not a retry") + }) + + // First attempt running: no pending next dispatch, and no retry interval reported while running. + t.Run("FirstAttemptRunning", func(t *testing.T) { + both(t, activityInfoProjection{ + State: enumspb.PENDING_ACTIVITY_STATE_STARTED, + Attempt: 1, + }, func(d retryDriver, t *testing.T) activityInfoProjection { + return d.start_Poll_ObserveRunning(t) + }) + }) + + // Backing off before the retry is dispatched: both the interval and the next-attempt schedule time + // are populated. + t.Run("BackingOff", func(t *testing.T) { + both(t, activityInfoProjection{ + State: enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, + Attempt: 2, + CurrentRetryInterval: backingOffInterval, + NextAttemptScheduleSet: true, + }, func(d retryDriver, t *testing.T) activityInfoProjection { + return d.start_Poll_FailRetryably_ObserveBackingOff(t) + }) + }) + + // Backing off after a worker-supplied next_retry_delay: the reported interval is the worker's + // override. + t.Run("NextRetryDelayOverride", func(t *testing.T) { + both(t, activityInfoProjection{ + State: enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, + Attempt: 2, + CurrentRetryInterval: nextRetryDelayOverride, + NextAttemptScheduleSet: true, + }, func(d retryDriver, t *testing.T) activityInfoProjection { + return d.start_Poll_FailWithNextRetryDelay_ObserveBackingOff(t, nextRetryDelayOverride) + }) + }) + + // Retry dispatched to Matching but not yet polled: both fields are nil. + t.Run("RetryDispatched", func(t *testing.T) { + both(t, activityInfoProjection{ + State: enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, + Attempt: 2, + }, func(d retryDriver, t *testing.T) activityInfoProjection { + return d.start_Poll_FailRetryably_RetryDispatched(t) + }) + }) + + // Retry attempt running with a further retry still permitted (max 3): nothing pending while running. + t.Run("RetryAttemptRunning", func(t *testing.T) { + both(t, activityInfoProjection{ + State: enumspb.PENDING_ACTIVITY_STATE_STARTED, + Attempt: 2, + }, func(d retryDriver, t *testing.T) activityInfoProjection { + return d.start_Poll_FailRetryably_BackoffElapses_Poll_ObserveRunning(t, 3) + }) + }) + + // Final attempt running with no retry remaining (max 2): still nothing pending while running. + t.Run("FinalAttemptRunning", func(t *testing.T) { + both(t, activityInfoProjection{ + State: enumspb.PENDING_ACTIVITY_STATE_STARTED, + Attempt: 2, + }, func(d retryDriver, t *testing.T) activityInfoProjection { + return d.start_Poll_FailRetryably_BackoffElapses_Poll_ObserveRunning(t, 2) + }) + }) + + // Paused while still backing off: dispatch will not occur while paused, so neither the interval nor + // the next-attempt schedule time should be reported. + t.Run("PausedBeforeDispatch", func(t *testing.T) { + both(t, activityInfoProjection{ + State: enumspb.PENDING_ACTIVITY_STATE_PAUSED, + Attempt: 2, + }, func(d retryDriver, t *testing.T) activityInfoProjection { + return d.start_Poll_FailRetryably_Paused(t) + }) + }) + + // Paused after the retry was dispatched: the dispatched code path already nils both fields, and the + // pause preserves that. + t.Run("PausedAfterDispatch", func(t *testing.T) { + both(t, activityInfoProjection{ + State: enumspb.PENDING_ACTIVITY_STATE_PAUSED, + Attempt: 2, + }, func(d retryDriver, t *testing.T) activityInfoProjection { + return d.start_Poll_FailRetryably_BackoffElapses_Paused(t) + }) + }) +} + +// retryDriver drives an activity to a retry-backoff state and reports its public retry-scheduling info. +type retryDriver interface { + start_Poll_ObserveRunning(t *testing.T) activityInfoProjection + start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) activityInfoProjection + start_Poll_FailWithNextRetryDelay_ObserveBackingOff(t *testing.T, nextRetryDelay time.Duration) activityInfoProjection + start_Poll_FailRetryably_RetryDispatched(t *testing.T) activityInfoProjection + start_Poll_FailRetryably_BackoffElapses_Poll_ObserveRunning(t *testing.T, maxAttempts int32) activityInfoProjection + start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) activityInfoProjection + start_Poll_FailRetryably_Paused(t *testing.T) activityInfoProjection +} + +// activityInfoProjection is the slice of an activity's public info this suite compares across the two +// surfaces: run state, attempt, and the retry-scheduling metadata. CurrentRetryInterval is rounded to +// the second (WFA derives it by subtracting two stored timestamps; SAA stores it exactly). +// NextAttemptScheduleTime is compared by set-ness, since its absolute value differs run to run. +type activityInfoProjection struct { + State enumspb.PendingActivityState + Attempt int32 + CurrentRetryInterval time.Duration + NextAttemptScheduleSet bool +} + +// retryDispatched reports that the retry has been dispatched to Matching (attempt 2, scheduled, with no +// pending future dispatch) — the state reached once the backoff elapses. +func retryDispatched(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED && p.Attempt == 2 && !p.NextAttemptScheduleSet +} + +const ( + // backingOffInterval is long enough to observe an activity while it is still backing off. + backingOffInterval = 30 * time.Second + // dispatchInterval is short enough that the backoff elapses and the retry dispatches within the test. + dispatchInterval = 1 * time.Second + // nextRetryDelayOverride is a worker-supplied next_retry_delay, distinct from backingOffInterval so + // the reported interval cannot be confused with the policy's. + nextRetryDelayOverride = 10 * time.Second + // startDelay keeps a first attempt pending dispatch for the whole test. + startDelay = time.Hour +) + // --- standalone-activity driver ------------------------------------------------------------ // saaDriver drives one standalone activity through the frontend RPCs. @@ -145,8 +304,9 @@ func (d *saaDriver) pollTask(t *testing.T) { d.token = resp.GetTaskToken() } -// describeActivity returns the DescribeActivityExecution response. -func (d *saaDriver) describeActivity(t *testing.T) *workflowservice.DescribeActivityExecutionResponse { +// describeActivity returns the DescribeActivityExecution response. Takes require.TestingT so it can be +// driven either by the test's *testing.T or by an *await.T inside an await.Require poll. +func (d *saaDriver) describeActivity(t require.TestingT) *workflowservice.DescribeActivityExecutionResponse { resp, err := d.env.FrontendClient().DescribeActivityExecution(d.s.Context(), &workflowservice.DescribeActivityExecutionRequest{ Namespace: d.env.Namespace().String(), ActivityId: d.activityID, @@ -156,6 +316,142 @@ func (d *saaDriver) describeActivity(t *testing.T) *workflowservice.DescribeActi return resp } +func (d *saaDriver) start_Poll_ObserveRunning(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval, 3) + d.pollTask(t) // returns only once the start is recorded, so a single describe already sees STARTED + return d.observe(t) +} + +func (d *saaDriver) start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval, 3) + d.pollTask(t) + d.failRetryably(t) // synchronous: the reschedule to backing off is committed before it returns + return d.observe(t) +} + +func (d *saaDriver) start_Poll_FailWithNextRetryDelay_ObserveBackingOff(t *testing.T, nextRetryDelay time.Duration) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval, 3) + d.pollTask(t) + d.failWithNextRetryDelay(t, nextRetryDelay) // synchronous: the reschedule is committed before it returns + return d.observe(t) +} + +func (d *saaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval, 3) + d.pollTask(t) + d.failRetryably(t) + return d.awaitObserve(t, retryDispatched) +} + +func (d *saaDriver) start_Poll_FailRetryably_BackoffElapses_Poll_ObserveRunning(t *testing.T, maxAttempts int32) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval, maxAttempts) + d.pollTask(t) + d.failRetryably(t) + d.pollTask(t) // long-poll: blocks until the retry dispatches and the next attempt's start is recorded + return d.observe(t) +} + +func (d *saaDriver) start_Poll_FailRetryably_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval, 3) // long, so the pause lands well before the retry dispatches + d.pollTask(t) + d.failRetryably(t) // synchronous: rescheduled to backing off before it returns + d.pauseActivity(t) // synchronous: paused before it returns + return d.observe(t) // so a single describe sees the paused, backing-off activity +} + +func (d *saaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval, 3) + d.pollTask(t) + d.failRetryably(t) + d.awaitObserve(t, retryDispatched) // projection poll: the backoff elapsing bumps no version to long-poll on + d.pauseActivity(t) // synchronous + return d.observe(t) +} + +// startRequest builds a start request for an activity whose failures are retryable, with a constant +// backoff of retryInterval. +func (d *saaDriver) startRequest(id string, retryInterval time.Duration, maxAttempts int32) *workflowservice.StartActivityExecutionRequest { + return &workflowservice.StartActivityExecutionRequest{ + Namespace: d.env.Namespace().String(), + ActivityId: id, + ActivityType: d.env.Tv().ActivityType(), + Identity: "worker", + Input: defaultInput, + TaskQueue: &taskqueuepb.TaskQueue{Name: id}, + StartToCloseTimeout: durationpb.New(time.Hour), + RetryPolicy: &commonpb.RetryPolicy{ + InitialInterval: durationpb.New(retryInterval), + BackoffCoefficient: 1.0, + MaximumInterval: durationpb.New(retryInterval), + MaximumAttempts: maxAttempts, + }, + RequestId: uuid.NewString(), + } +} + +func (d *saaDriver) startRetryable(t *testing.T, retryInterval time.Duration, maxAttempts int32) { + id := testcore.RandomizeStr(t.Name()) + resp, err := d.env.FrontendClient().StartActivityExecution(d.s.Context(), d.startRequest(id, retryInterval, maxAttempts)) + require.NoError(t, err) + d.activityID, d.taskQueue, d.runID = id, id, resp.RunId +} + +// startWithStartDelay starts an activity with a start delay so the first attempt stays pending dispatch. +func (d *saaDriver) startWithStartDelay(t *testing.T, startDelay time.Duration) { + id := testcore.RandomizeStr(t.Name()) + req := d.startRequest(id, backingOffInterval, 3) + req.StartDelay = durationpb.New(startDelay) + resp, err := d.env.FrontendClient().StartActivityExecution(d.s.Context(), req) + require.NoError(t, err) + d.activityID, d.taskQueue, d.runID = id, id, resp.RunId +} + +func (d *saaDriver) failRetryably(t *testing.T) { + _, err := d.env.FrontendClient().RespondActivityTaskFailed(d.s.Context(), &workflowservice.RespondActivityTaskFailedRequest{ + Namespace: d.env.Namespace().String(), TaskToken: d.token, Identity: "worker", Failure: retryableFailure(), + }) + require.NoError(t, err) +} + +func (d *saaDriver) failWithNextRetryDelay(t *testing.T, nextRetryDelay time.Duration) { + _, err := d.env.FrontendClient().RespondActivityTaskFailed(d.s.Context(), &workflowservice.RespondActivityTaskFailedRequest{ + Namespace: d.env.Namespace().String(), TaskToken: d.token, Identity: "worker", + Failure: retryableFailureWithNextRetryDelay(nextRetryDelay), + }) + require.NoError(t, err) +} + +func (d *saaDriver) pauseActivity(t *testing.T) { + _, err := d.env.FrontendClient().PauseActivityExecution(d.s.Context(), &workflowservice.PauseActivityExecutionRequest{ + Namespace: d.env.Namespace().String(), ActivityId: d.activityID, RunId: d.runID, Identity: "op", Reason: "drive", RequestId: uuid.NewString(), + }) + require.NoError(t, err) +} + +// observe reads the activity's public retry-scheduling info as the shared projection. +func (d *saaDriver) observe(t require.TestingT) activityInfoProjection { + i := d.describeActivity(t).GetInfo() + return activityInfoProjection{ + State: i.GetRunState(), + Attempt: i.GetAttempt(), + CurrentRetryInterval: i.GetCurrentRetryInterval().AsDuration().Round(time.Second), + NextAttemptScheduleSet: i.GetNextAttemptScheduleTime() != nil, + } +} + +// awaitObserve polls the public projection until pred holds, returning that projection. Reserved for a +// dispatch-delay elapse (start-delay / backoff), whose only effect is a read-time projection flip with no +// transition-history version advance, so a Describe long-poll would never wake. Effects committed by a +// synchronous RPC (or already recorded by the time a Poll returns) need no wait — describe directly. +func (d *saaDriver) awaitObserve(t *testing.T, pred func(activityInfoProjection) bool) activityInfoProjection { + var p activityInfoProjection + await.Require(d.s.Context(), t, func(at *await.T) { + p = d.observe(at) + at.Require().Truef(pred(p), "last observed: %+v", p) + }, 15*time.Second, 100*time.Millisecond) + return p +} + // --- workflow-activity driver -------------------------------------------------------------- // wfaDriver drives one activity scheduled by a helper workflow. @@ -194,6 +490,7 @@ func (d *wfaDriver) startWithNonRetryableTimeout(t *testing.T, timeoutType enums p := workflowActivityParams{ TaskQueue: d.activityTQ, StartToClose: time.Hour, + RetryInterval: 200 * time.Millisecond, MaxAttempts: 3, NonRetryableErrorTypes: []string{retrypolicy.TimeoutFailureTypePrefix + timeoutType.String()}, } @@ -247,11 +544,142 @@ func (d *wfaDriver) awaitTerminalStatus(t *testing.T) enumspb.ActivityExecutionS } } +func (d *wfaDriver) start_Poll_ObserveRunning(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval, 3) + d.pollTask(t) // returns only once the start is recorded, so a single describe already sees STARTED + return d.observe(t) +} + +func (d *wfaDriver) start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval, 3) + d.pollTask(t) + d.failRetryably(t) // synchronous: the reschedule to backing off is committed before it returns + return d.observe(t) +} + +func (d *wfaDriver) start_Poll_FailWithNextRetryDelay_ObserveBackingOff(t *testing.T, nextRetryDelay time.Duration) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval, 3) + d.pollTask(t) + d.failWithNextRetryDelay(t, nextRetryDelay) // synchronous: the reschedule is committed before it returns + return d.observe(t) +} + +func (d *wfaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval, 3) + d.pollTask(t) + d.failRetryably(t) + return d.awaitObserve(t, retryDispatched) +} + +func (d *wfaDriver) start_Poll_FailRetryably_BackoffElapses_Poll_ObserveRunning(t *testing.T, maxAttempts int32) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval, maxAttempts) + d.pollTask(t) + d.failRetryably(t) + d.pollTask(t) // long-poll: blocks until the retry dispatches and the next attempt's start is recorded + return d.observe(t) +} + +func (d *wfaDriver) start_Poll_FailRetryably_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval, 3) // long, so the pause lands well before the retry dispatches + d.pollTask(t) + d.failRetryably(t) // synchronous: rescheduled to backing off before it returns + d.pauseActivity(t) // synchronous: paused before it returns + return d.observe(t) // so a single describe sees the paused, backing-off activity +} + +func (d *wfaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval, 3) + d.pollTask(t) + d.failRetryably(t) + d.awaitObserve(t, retryDispatched) // projection poll: the backoff elapsing bumps no version to long-poll on + d.pauseActivity(t) // synchronous + return d.observe(t) +} + +// startRetryable starts a workflow that schedules one activity whose failures are retryable, with a +// constant backoff of retryInterval. +func (d *wfaDriver) startRetryable(t *testing.T, retryInterval time.Duration, maxAttempts int32) { + wfTQ := testcore.RandomizeStr("parity-wf") + d.activityTQ = testcore.RandomizeStr("parity-act") + + w := sdkworker.New(d.env.SdkClient(), wfTQ, sdkworker.Options{}) + w.RegisterWorkflow(singleActivityWorkflow) + require.NoError(t, w.Start()) + t.Cleanup(w.Stop) + + run, err := d.env.SdkClient().ExecuteWorkflow(d.s.Context(), + sdkclient.StartWorkflowOptions{ID: testcore.RandomizeStr("parity-run"), TaskQueue: wfTQ}, + singleActivityWorkflow, workflowActivityParams{ + TaskQueue: d.activityTQ, StartToClose: time.Hour, RetryInterval: retryInterval, MaxAttempts: maxAttempts, + }) + require.NoError(t, err) + d.run = run +} + +func (d *wfaDriver) failRetryably(t *testing.T) { + _, err := d.env.FrontendClient().RespondActivityTaskFailed(d.s.Context(), &workflowservice.RespondActivityTaskFailedRequest{ + Namespace: d.env.Namespace().String(), TaskToken: d.token, Identity: "worker", Failure: retryableFailure(), + }) + require.NoError(t, err) +} + +func (d *wfaDriver) failWithNextRetryDelay(t *testing.T, nextRetryDelay time.Duration) { + _, err := d.env.FrontendClient().RespondActivityTaskFailed(d.s.Context(), &workflowservice.RespondActivityTaskFailedRequest{ + Namespace: d.env.Namespace().String(), TaskToken: d.token, Identity: "worker", + Failure: retryableFailureWithNextRetryDelay(nextRetryDelay), + }) + require.NoError(t, err) +} + +func (d *wfaDriver) pauseActivity(t *testing.T) { + _, err := d.env.FrontendClient().PauseActivityExecution(d.s.Context(), &workflowservice.PauseActivityExecutionRequest{ + Namespace: d.env.Namespace().String(), WorkflowId: d.run.GetID(), RunId: d.run.GetRunID(), ActivityId: wfaActivityID, + Identity: "op", Reason: "drive", RequestId: uuid.NewString(), + }) + require.NoError(t, err) +} + +// observe reads the activity's public retry-scheduling info via DescribeWorkflowExecution, as the shared +// projection. Takes require.TestingT so it works under an await.Require poll (see awaitObserve). +func (d *wfaDriver) observe(t require.TestingT) activityInfoProjection { + resp, err := d.env.SdkClient().DescribeWorkflowExecution(d.s.Context(), d.run.GetID(), d.run.GetRunID()) + require.NoError(t, err) + for _, pa := range resp.GetPendingActivities() { + if pa.GetActivityId() == wfaActivityID { + return activityInfoProjection{ + State: pa.GetState(), + Attempt: pa.GetAttempt(), + CurrentRetryInterval: pa.GetCurrentRetryInterval().AsDuration().Round(time.Second), + NextAttemptScheduleSet: pa.GetNextAttemptScheduleTime() != nil, + } + } + } + require.Fail(t, "no pending activity", "activity %q not pending; workflow may have closed", wfaActivityID) + return activityInfoProjection{} +} + +// awaitObserve polls the public projection until pred holds, returning that projection. Reserved for a +// dispatch-delay elapse (start-delay / backoff), whose only effect is a read-time projection flip with no +// transition-history version advance, so a Describe long-poll would never wake. Effects committed by a +// synchronous RPC (or already recorded by the time a Poll returns) need no wait — describe directly. +func (d *wfaDriver) awaitObserve(t *testing.T, pred func(activityInfoProjection) bool) activityInfoProjection { + var p activityInfoProjection + await.Require(d.s.Context(), t, func(at *await.T) { + p = d.observe(at) + at.Require().Truef(pred(p), "last observed: %+v", p) + }, 15*time.Second, 100*time.Millisecond) + return p +} + +// wfaActivityID is the fixed ID the helper workflow assigns its activity. +const wfaActivityID = "act" + // workflowActivityParams configures the single activity the helper workflow schedules. type workflowActivityParams struct { TaskQueue string StartToClose time.Duration Heartbeat time.Duration // 0 = unset + RetryInterval time.Duration // InitialInterval == MaximumInterval (constant backoff) MaxAttempts int32 NonRetryableErrorTypes []string } @@ -260,16 +688,37 @@ type workflowActivityParams struct { func singleActivityWorkflow(ctx workflow.Context, p workflowActivityParams) error { ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ TaskQueue: p.TaskQueue, - ActivityID: "act", + ActivityID: wfaActivityID, StartToCloseTimeout: p.StartToClose, HeartbeatTimeout: p.Heartbeat, RetryPolicy: &temporal.RetryPolicy{ - InitialInterval: 200 * time.Millisecond, + InitialInterval: p.RetryInterval, BackoffCoefficient: 1.0, - MaximumInterval: 200 * time.Millisecond, + MaximumInterval: p.RetryInterval, MaximumAttempts: p.MaxAttempts, NonRetryableErrorTypes: p.NonRetryableErrorTypes, }, }) return workflow.ExecuteActivity(ctx, "noopActivity").Get(ctx, nil) } + +// retryableFailure is the worker-reported failure the drivers use to trigger a retry. +func retryableFailure() *failurepb.Failure { + return &failurepb.Failure{ + Message: "drive", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{Type: "drive"}, + }, + } +} + +// retryableFailureWithNextRetryDelay is a retryable failure carrying a worker-supplied next_retry_delay +// that overrides the policy backoff for the next attempt. +func retryableFailureWithNextRetryDelay(nextRetryDelay time.Duration) *failurepb.Failure { + return &failurepb.Failure{ + Message: "drive", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{Type: "drive", NextRetryDelay: durationpb.New(nextRetryDelay)}, + }, + } +} diff --git a/tests/activity_standalone_test.go b/tests/activity_standalone_test.go index c7c1b08b93e..87604a616a1 100644 --- a/tests/activity_standalone_test.go +++ b/tests/activity_standalone_test.go @@ -14502,201 +14502,3 @@ func (s *standaloneActivityTestSuite) TestResetActivityExecution() { require.NoError(t, err) }) } - -func (s *standaloneActivityTestSuite) TestNextAttemptScheduleTimeAndCurrentRetryInterval() { - const ( - retryInterval = 5 * time.Second // long enough to observe a backoff window before the retry dispatches - backoffSettle = 2 * time.Second // slack so a wait outlasts the backoff's firing instant - startDelay = time.Hour // keeps the first attempt pending dispatch for the whole test - ) - - startActivity := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv, maxAttempts int32, startDelay time.Duration) (activityID, taskQueue string) { - activityID = testcore.RandomizeStr(t.Name()) - taskQueue = testcore.RandomizeStr(t.Name()) - req := &workflowservice.StartActivityExecutionRequest{ - Namespace: env.Namespace().String(), - ActivityId: activityID, - ActivityType: env.Tv().ActivityType(), - Identity: defaultIdentity, - Input: defaultInput, - TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, - StartToCloseTimeout: durationpb.New(time.Hour), - RetryPolicy: &commonpb.RetryPolicy{ - InitialInterval: durationpb.New(retryInterval), - BackoffCoefficient: 2.0, // long enough to observe during backoff window - MaximumInterval: durationpb.New(retryInterval), - MaximumAttempts: maxAttempts, - }, - } - if startDelay > 0 { - req.StartDelay = durationpb.New(startDelay) - } - _, err := env.FrontendClient().StartActivityExecution(s.Context(), req) - require.NoError(t, err) - return activityID, taskQueue - } - - pollTask := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv, taskQueue string) []byte { - resp, err := env.pollActivityTaskQueue(s.Context(), taskQueue) - require.NoError(t, err) - return resp.GetTaskToken() - } - - respondFailedRetryably := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv, token []byte) { - _, err := env.FrontendClient().RespondActivityTaskFailed(s.Context(), &workflowservice.RespondActivityTaskFailedRequest{ - Namespace: env.Namespace().String(), - TaskToken: token, - Failure: &failurepb.Failure{ - Message: "drive", - FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{Type: "drive", NonRetryable: false}}, - }, - }) - require.NoError(t, err) - } - - describeActivity := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv, activityID string) *activitypb.ActivityExecutionInfo { - resp, err := env.FrontendClient().DescribeActivityExecution(s.Context(), &workflowservice.DescribeActivityExecutionRequest{ - Namespace: env.Namespace().String(), - ActivityId: activityID, - }) - require.NoError(t, err) - return resp.GetInfo() - } - - startWithStartDelay := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv) *activitypb.ActivityExecutionInfo { - activityID, _ := startActivity(s, t, env, 0, startDelay) - return describeActivity(s, t, env, activityID) - } - - // The driver closures below are named for the sequence of steps they run, delimited by - // underscores; ST1003 is silenced on each declaration to keep those names. - start_Poll := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv, maxAttempts int32) *activitypb.ActivityExecutionInfo { //nolint:staticcheck // ST1003: underscores delimit the driver step sequence - activityID, taskQueue := startActivity(s, t, env, maxAttempts, 0) - pollTask(s, t, env, taskQueue) - return describeActivity(s, t, env, activityID) - } - - start_Poll_FailRetryably := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv, maxAttempts int32) *activitypb.ActivityExecutionInfo { //nolint:staticcheck // ST1003: underscores delimit the driver step sequence - activityID, taskQueue := startActivity(s, t, env, maxAttempts, 0) - token := pollTask(s, t, env, taskQueue) - respondFailedRetryably(s, t, env, token) - return describeActivity(s, t, env, activityID) - } - - start_Poll_FailRetryably_RetryBackoffElapse := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv, maxAttempts int32) *activitypb.ActivityExecutionInfo { //nolint:staticcheck // ST1003: underscores delimit the driver step sequence - activityID, taskQueue := startActivity(s, t, env, maxAttempts, 0) - token := pollTask(s, t, env, taskQueue) - respondFailedRetryably(s, t, env, token) - time.Sleep(retryInterval + backoffSettle) //nolint:forbidigo - return describeActivity(s, t, env, activityID) - } - - start_Poll_FailRetryably_RetryBackoffElapse_Poll := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv, maxAttempts int32) *activitypb.ActivityExecutionInfo { //nolint:staticcheck // ST1003: underscores delimit the driver step sequence - activityID, taskQueue := startActivity(s, t, env, maxAttempts, 0) - token := pollTask(s, t, env, taskQueue) - respondFailedRetryably(s, t, env, token) - time.Sleep(retryInterval + backoffSettle) //nolint:forbidigo - pollTask(s, t, env, taskQueue) - return describeActivity(s, t, env, activityID) - } - - start_Poll_FailWithNextRetryDelay := func(s *standaloneActivityTestSuite, t *testing.T, env *standaloneActivityEnv, maxAttempts int32, nextRetryDelay time.Duration) *activitypb.ActivityExecutionInfo { //nolint:staticcheck // ST1003: underscores delimit the driver step sequence - activityID, taskQueue := startActivity(s, t, env, maxAttempts, 0) - token := pollTask(s, t, env, taskQueue) - _, err := env.FrontendClient().RespondActivityTaskFailed(s.Context(), &workflowservice.RespondActivityTaskFailedRequest{ - Namespace: env.Namespace().String(), - TaskToken: token, - Failure: &failurepb.Failure{ - Message: "drive", - FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ - Type: "drive", NonRetryable: false, NextRetryDelay: durationpb.New(nextRetryDelay), - }}, - }, - }) - require.NoError(t, err) - return describeActivity(s, t, env, activityID) - } - - // First attempt within its start delay: the dispatch is pending in the future, and this is not a retry. - s.Run("StartDelayPending", func(s *standaloneActivityTestSuite) { - t := s.T() - env := s.newTestEnv() - - info := startWithStartDelay(s, t, env) - require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, info.GetRunState()) - require.Equal(t, info.GetExecutionTime().AsTime(), info.GetNextAttemptScheduleTime().AsTime(), - "during a start delay, NextAttemptScheduleTime is the pending dispatch time (schedule+delay)") - require.Nil(t, info.GetCurrentRetryInterval(), "the first attempt is not a retry") - }) - - // First attempt running: no pending next dispatch, and no retry interval reported while running. - s.Run("FirstAttemptRunning", func(s *standaloneActivityTestSuite) { - t := s.T() - env := s.newTestEnv() - - info := start_Poll(s, t, env, 3) - require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_STARTED, info.GetRunState()) - require.Nil(t, info.GetNextAttemptScheduleTime(), "null while running") - require.Nil(t, info.GetCurrentRetryInterval(), "null while running") - }) - - // Backing off before the retry dispatches: the next dispatch is in the future. - s.Run("BackingOffBeforeDispatch", func(s *standaloneActivityTestSuite) { - t := s.T() - env := s.newTestEnv() - - info := start_Poll_FailRetryably(s, t, env, 3) - require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, info.GetRunState()) - require.True(t, info.GetNextAttemptScheduleTime().AsTime().After(time.Now()), "future retry dispatch time") - require.Equal(t, retryInterval, info.GetCurrentRetryInterval().AsDuration(), "while backing off, the current interval") - }) - - // Retry queued in matching but not yet started. Both next_attempt_schedule_time and - // current_retry_interval are null. - s.Run("BackingOffAfterDispatch", func(s *standaloneActivityTestSuite) { - t := s.T() - env := s.newTestEnv() - info := start_Poll_FailRetryably_RetryBackoffElapse(s, t, env, 3) - require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, info.GetRunState()) - require.EqualValues(t, 2, info.GetAttempt()) - require.Nil(t, info.GetNextAttemptScheduleTime()) - require.Nil(t, info.GetCurrentRetryInterval()) - }) - // Backing off after a worker-supplied next_retry_delay: the reported interval is the worker's - // override, not the policy's InitialInterval. - s.Run("BackingOffAfterNextRetryDelayOverride", func(s *standaloneActivityTestSuite) { - t := s.T() - env := s.newTestEnv() - - const override = 10 * time.Second // distinct from the policy interval so the two can't be confused - info := start_Poll_FailWithNextRetryDelay(s, t, env, 3, override) - require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, info.GetRunState()) - require.True(t, info.GetNextAttemptScheduleTime().AsTime().After(time.Now()), "future retry dispatch time") - require.Equal(t, override, info.GetCurrentRetryInterval().AsDuration(), - "while backing off, the current interval is the worker's next_retry_delay override") - }) - - // Retry attempt running with a further retry permitted. - s.Run("RetryAttemptRunning", func(s *standaloneActivityTestSuite) { - t := s.T() - env := s.newTestEnv() - - info := start_Poll_FailRetryably_RetryBackoffElapse_Poll(s, t, env, 3) - require.EqualValues(t, 2, info.GetAttempt()) - require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_STARTED, info.GetRunState()) - require.Nil(t, info.GetNextAttemptScheduleTime(), "null while running") - require.Nil(t, info.GetCurrentRetryInterval(), "null while running") - }) - - // Final attempt running with no retry remaining. - s.Run("FinalAttemptRunning", func(s *standaloneActivityTestSuite) { - t := s.T() - env := s.newTestEnv() - - info := start_Poll_FailRetryably_RetryBackoffElapse_Poll(s, t, env, 2) - require.EqualValues(t, 2, info.GetAttempt()) - require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_STARTED, info.GetRunState()) - require.Nil(t, info.GetNextAttemptScheduleTime(), "null while running") - require.Nil(t, info.GetCurrentRetryInterval(), "null while running") - }) -} diff --git a/tests/xdc/activity_api_test.go b/tests/xdc/activity_api_test.go index f63d952af18..97b866f9a43 100644 --- a/tests/xdc/activity_api_test.go +++ b/tests/xdc/activity_api_test.go @@ -154,7 +154,7 @@ func (s *ActivityApiStateReplicationSuite) TestPauseActivityFailover() { if description.GetPendingActivities() != nil { require.Len(t, description.PendingActivities, 1) require.True(t, description.PendingActivities[0].Paused) - require.Equal(t, int64(2), description.PendingActivities[0].CurrentRetryInterval.GetSeconds()) + require.Nil(t, description.PendingActivities[0].CurrentRetryInterval) } }, 5*time.Second, 200*time.Millisecond) @@ -180,7 +180,7 @@ func (s *ActivityApiStateReplicationSuite) TestPauseActivityFailover() { require.Len(t, description.PendingActivities, 1) require.True(t, description.PendingActivities[0].Paused) require.Equal(t, int32(1), description.PendingActivities[0].Attempt) - require.Equal(t, int64(2), description.PendingActivities[0].CurrentRetryInterval.GetSeconds()) + require.Nil(t, description.PendingActivities[0].CurrentRetryInterval) } }, 5*time.Second, 200*time.Millisecond) @@ -206,7 +206,7 @@ func (s *ActivityApiStateReplicationSuite) TestPauseActivityFailover() { if description.GetPendingActivities() != nil { require.Len(t, description.PendingActivities, 1) require.True(t, description.PendingActivities[0].Paused) - require.Equal(t, int64(2), description.PendingActivities[0].CurrentRetryInterval.GetSeconds()) + require.Nil(t, description.PendingActivities[0].CurrentRetryInterval) require.Equal(t, int32(10), description.PendingActivities[0].MaximumAttempts) } }, 5*time.Second, 200*time.Millisecond)