From c86aa47b21434da34c10e57032e6e80ed1c3b6df Mon Sep 17 00:00:00 2001 From: Dan Davison Date: Tue, 21 Jul 2026 21:18:10 -0400 Subject: [PATCH 1/9] Test current_retry_interval is nil after dispatch to Matching Covers the SCHEDULED branch where the retry has already been dispatched to Matching (now >= scheduledTime): both NextAttemptScheduleTime and CurrentRetryInterval must be nil. Run: go test ./service/history/workflow/ -run 'TestActivitySuite/TestGetPendingActivityInfoRetryDispatchedToMatching' -count=1 --- service/history/workflow/activity_test.go | 30 +++++++++++++++++++++++ 1 file changed, 30 insertions(+) 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{ From c5b71959d5dbe98c0c8db09ab47d965b056fc0b1 Mon Sep 17 00:00:00 2001 From: Dan Davison Date: Tue, 21 Jul 2026 21:08:17 -0400 Subject: [PATCH 2/9] Breaking change to current_retry_interval Return nil after dispatch to Matching. Consider an activity that's in retry backoff before attempt 2. This retry interval is 5s and the next one, if there is another, will be 10s. current_retry_interval: SCHEDULED (dispatched to matching) STARTED (attempt completed/failed) before 5s 10s nil after 5s nil nil next_attempt_schedule_time for comparison: SCHEDULED (dispatched to matching at t) STARTED (attempt completed/failed) t nil nil --- service/history/workflow/activity.go | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/service/history/workflow/activity.go b/service/history/workflow/activity.go index 12a0f984fd6..6e03ae9e144 100644 --- a/service/history/workflow/activity.go +++ b/service/history/workflow/activity.go @@ -121,16 +121,14 @@ func GetPendingActivityInfo( 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 + // 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 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 } } } From 99743dc9e67dfa6eed7021ba7c155a47a83d0d4f Mon Sep 17 00:00:00 2001 From: Dan Davison Date: Thu, 23 Jul 2026 07:56:14 -0400 Subject: [PATCH 3/9] Expect nil current_retry_interval for paused activity A paused activity is held on the server and not dispatched to Matching, but it reaches the same code path as a dispatched retry, so current_retry_interval is now nil for it too. Run: go test ./tests/xdc/ -run 'TestActivityApiStateReplicationSuite/TestPauseActivityFailover' --- tests/xdc/activity_api_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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) From 4700086f46f9b92ce9d0a9b669271f98f3d1f3d1 Mon Sep 17 00:00:00 2001 From: Dan Davison Date: Thu, 23 Jul 2026 14:41:00 -0400 Subject: [PATCH 4/9] Parity tests: current_retry_interval across retry-backoff states Adds TestParityCurrentRetryInterval, driving both a workflow activity (the oracle) and a standalone activity through the same retry-backoff states and asserting the same public retry-scheduling info (current_retry_interval, next_attempt_schedule_time), via the halfway-house saaDriver/wfaDriver harness: - BackingOff: before the retry is dispatched to Matching, both fields are populated. - RetryDispatched: once dispatched (now >= scheduledTime), both are nil. - PausedAfterDispatch: pausing a dispatched retry preserves the nil. All three pass on both surfaces, so the current_retry_interval contract the WFA change locks holds for SAA too. Parameterizes the shared singleActivityWorkflow with RetryInterval. Run: go test -tags test_dep -run 'TestStandaloneActivityTestSuite/TestParityCurrentRetryInterval' -count=1 -v -timeout 480s ./tests/ --- tests/activity_parity_test.go | 291 +++++++++++++++++++++++++++++++++- 1 file changed, 288 insertions(+), 3 deletions(-) diff --git a/tests/activity_parity_test.go b/tests/activity_parity_test.go index 388e9b14cf3..3eb063d716f 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" @@ -59,6 +60,82 @@ 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 only while a retry is backing off +// (before it is dispatched to Matching). Once the retry is dispatched — or while the activity is paused, +// which reaches the same server code path — both are nil. WFA is the oracle; SAA must match. +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)) + }) + } + + // 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) + }) + }) + + // 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) + }) + }) + + // 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_FailRetryably_ObserveBackingOff(t *testing.T) activityInfoProjection + start_Poll_FailRetryably_RetryDispatched(t *testing.T) activityInfoProjection + start_Poll_FailRetryably_BackoffElapses_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 +} + +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 +) + // --- standalone-activity driver ------------------------------------------------------------ // saaDriver drives one standalone activity through the frontend RPCs. @@ -156,6 +233,100 @@ func (d *saaDriver) describeActivity(t *testing.T) *workflowservice.DescribeActi return resp } +func (d *saaDriver) start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval) + d.pollTask(t) + d.failRetryably(t) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED && p.Attempt == 2 + }) +} + +func (d *saaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval) + d.pollTask(t) + d.failRetryably(t) + d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) + time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching + return d.observe(t) +} + +func (d *saaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval) + d.pollTask(t) + d.failRetryably(t) + d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) + time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching + d.pauseActivity(t) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_PAUSED + }) +} + +// startRetryable starts an activity whose failures are retryable, with a constant backoff of retryInterval. +func (d *saaDriver) startRetryable(t *testing.T, retryInterval time.Duration) { + id := testcore.RandomizeStr(t.Name()) + resp, err := d.env.FrontendClient().StartActivityExecution(d.s.Context(), &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: 3, + }, + RequestId: uuid.NewString(), + }) + 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) 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 *testing.T) 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 observe until pred holds, returning that projection. +func (d *saaDriver) awaitObserve(t *testing.T, pred func(activityInfoProjection) bool) activityInfoProjection { + deadline := time.Now().Add(15 * time.Second) + for { + p := d.observe(t) + if pred(p) { + return p + } + if time.Now().After(deadline) { + require.Fail(t, "activity did not reach the expected state", "last observed: %+v", p) + return p + } + time.Sleep(100 * time.Millisecond) + } +} + // --- workflow-activity driver -------------------------------------------------------------- // wfaDriver drives one activity scheduled by a helper workflow. @@ -194,6 +365,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 +419,114 @@ func (d *wfaDriver) awaitTerminalStatus(t *testing.T) enumspb.ActivityExecutionS } } +func (d *wfaDriver) start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval) + d.pollTask(t) + d.failRetryably(t) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED && p.Attempt == 2 + }) +} + +func (d *wfaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval) + d.pollTask(t) + d.failRetryably(t) + d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) + time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching + return d.observe(t) +} + +func (d *wfaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, dispatchInterval) + d.pollTask(t) + d.failRetryably(t) + d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) + time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching + d.pauseActivity(t) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_PAUSED + }) +} + +// 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) { + 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: 3, + }) + 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) 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. +func (d *wfaDriver) observe(t *testing.T) 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 observe until pred holds, returning that projection. +func (d *wfaDriver) awaitObserve(t *testing.T, pred func(activityInfoProjection) bool) activityInfoProjection { + deadline := time.Now().Add(15 * time.Second) + for { + p := d.observe(t) + if pred(p) { + return p + } + if time.Now().After(deadline) { + require.Fail(t, "activity did not reach the expected state", "last observed: %+v", p) + return p + } + time.Sleep(100 * time.Millisecond) + } +} + +// 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 +535,26 @@ 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"}, + }, + } +} From 14df9d7721581d0b83ee1f9cca3dd107d3f10a83 Mon Sep 17 00:00:00 2001 From: Dan Davison Date: Thu, 23 Jul 2026 15:57:02 -0400 Subject: [PATCH 5/9] Parity test: current_retry_interval should be nil while paused before dispatch Adds PausedBeforeDispatch: an activity paused while still backing off should report neither current_retry_interval nor next_attempt_schedule_time, since no dispatch occurs while paused. This FAILS today: WFA keeps reporting the backing-off values until it flips the state to PAUSED, whereas SAA already nils them. The failing assertion records the intended end state. Run: go test -tags test_dep -run 'TestStandaloneActivityTestSuite/TestParityCurrentRetryInterval' -count=1 -v -timeout 480s ./tests/ --- tests/activity_parity_test.go | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/tests/activity_parity_test.go b/tests/activity_parity_test.go index 3eb063d716f..96d60bfdd75 100644 --- a/tests/activity_parity_test.go +++ b/tests/activity_parity_test.go @@ -99,6 +99,18 @@ func (s *standaloneActivityTestSuite) TestParityCurrentRetryInterval() { }) }) + // Paused while still backing off: dispatch will not occur while paused, so neither the interval nor + // the next-attempt schedule time should be reported. (Fails against WFA today, which keeps reporting + // the backing-off values until it flips the state to PAUSED — see start_Poll_FailRetryably_Paused.) + 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) { @@ -116,6 +128,7 @@ type retryDriver interface { start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) activityInfoProjection start_Poll_FailRetryably_RetryDispatched(t *testing.T) 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 @@ -251,6 +264,17 @@ func (d *saaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activ return d.observe(t) } +func (d *saaDriver) start_Poll_FailRetryably_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval) // long, so the pause and read land well before the retry dispatches + d.pollTask(t) + d.failRetryably(t) + d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) + d.pauseActivity(t) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_PAUSED + }) +} + func (d *saaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores d.startRetryable(t, dispatchInterval) d.pollTask(t) @@ -437,6 +461,17 @@ func (d *wfaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activ return d.observe(t) } +func (d *wfaDriver) start_Poll_FailRetryably_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores + d.startRetryable(t, backingOffInterval) // long, so the pause and read land well before the retry dispatches + d.pollTask(t) + d.failRetryably(t) + d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) + d.pauseActivity(t) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_PAUSED + }) +} + func (d *wfaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores d.startRetryable(t, dispatchInterval) d.pollTask(t) From dd3c00ca6f74d27c988e1759574136097221e922 Mon Sep 17 00:00:00 2001 From: Dan Davison Date: Thu, 23 Jul 2026 16:02:13 -0400 Subject: [PATCH 6/9] Report nil current_retry_interval for a paused, backing-off activity GetPendingActivityInfo populated next_attempt_schedule_time / current_retry_interval whenever a SCHEDULED activity was still before its scheduled retry time, even when paused. A paused activity will not dispatch, so it must report neither. Gate the backing-off branch on !ai.Paused. Fixes TestParityCurrentRetryInterval/PausedBeforeDispatch. Run: go test ./service/history/workflow/ -run TestActivitySuite -count=1 go test -tags test_dep -run 'TestStandaloneActivityTestSuite/TestParityCurrentRetryInterval' -count=1 -v -timeout 480s ./tests/ --- service/history/workflow/activity.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/service/history/workflow/activity.go b/service/history/workflow/activity.go index 6e03ae9e144..d29c6f9ad18 100644 --- a/service/history/workflow/activity.go +++ b/service/history/workflow/activity.go @@ -120,13 +120,13 @@ func GetPendingActivityInfo( p.Attempt = ai.Attempt if p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED { scheduledTime := ai.ScheduledTime.AsTime() - if now.Before(scheduledTime) { + 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 { - // retry has been dispatched to Matching + // retry has been dispatched to Matching, or the activity is paused so no dispatch will occur p.NextAttemptScheduleTime = nil p.CurrentRetryInterval = nil } From 34267ef11f48359d3406a401bd9af46b272695b3 Mon Sep 17 00:00:00 2001 From: Dan Davison Date: Thu, 23 Jul 2026 16:18:14 -0400 Subject: [PATCH 7/9] Fold the SAA-only retry-scheduling test into the parity test TestParityCurrentRetryInterval now covers everything the SAA-only TestNextAttemptScheduleTimeAndCurrentRetryInterval did, checking WFA (the oracle) and SAA together: FirstAttemptRunning, NextRetryDelayOverride, RetryAttemptRunning, and FinalAttemptRunning (max 2). StartDelayPending stays SAA-only (WFA has no per-activity start delay). Deletes the now-redundant SAA-only test. Run: go test -tags test_dep -run 'TestStandaloneActivityTestSuite/TestParityCurrentRetryInterval' -count=1 -v -timeout 580s ./tests/ --- tests/activity_parity_test.go | 204 +++++++++++++++++++++++++++--- tests/activity_standalone_test.go | 198 ----------------------------- 2 files changed, 183 insertions(+), 219 deletions(-) diff --git a/tests/activity_parity_test.go b/tests/activity_parity_test.go index 96d60bfdd75..5e544c6ce2c 100644 --- a/tests/activity_parity_test.go +++ b/tests/activity_parity_test.go @@ -60,9 +60,9 @@ 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 only while a retry is backing off -// (before it is dispatched to Matching). Once the retry is dispatched — or while the activity is paused, -// which reaches the same server code path — both are nil. WFA is the oracle; SAA must match. +// 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() @@ -76,6 +76,28 @@ func (s *standaloneActivityTestSuite) TestParityCurrentRetryInterval() { }) } + // 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) { @@ -89,6 +111,19 @@ func (s *standaloneActivityTestSuite) TestParityCurrentRetryInterval() { }) }) + // 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{ @@ -99,9 +134,28 @@ func (s *standaloneActivityTestSuite) TestParityCurrentRetryInterval() { }) }) + // 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. (Fails against WFA today, which keeps reporting - // the backing-off values until it flips the state to PAUSED — see start_Poll_FailRetryably_Paused.) + // the next-attempt schedule time should be reported. t.Run("PausedBeforeDispatch", func(t *testing.T) { both(t, activityInfoProjection{ State: enumspb.PENDING_ACTIVITY_STATE_PAUSED, @@ -125,8 +179,11 @@ func (s *standaloneActivityTestSuite) TestParityCurrentRetryInterval() { // 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 } @@ -147,6 +204,11 @@ const ( 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 ------------------------------------------------------------ @@ -246,8 +308,16 @@ 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) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_STARTED && p.Attempt == 1 + }) +} + func (d *saaDriver) start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores - d.startRetryable(t, backingOffInterval) + d.startRetryable(t, backingOffInterval, 3) d.pollTask(t) d.failRetryably(t) return d.awaitObserve(t, func(p activityInfoProjection) bool { @@ -255,8 +325,17 @@ func (d *saaDriver) start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) act }) } +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) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED && p.Attempt == 2 + }) +} + func (d *saaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores - d.startRetryable(t, dispatchInterval) + d.startRetryable(t, dispatchInterval, 3) d.pollTask(t) d.failRetryably(t) d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) @@ -264,8 +343,20 @@ func (d *saaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activ return d.observe(t) } +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.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) + time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching + d.pollTask(t) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_STARTED && p.Attempt == 2 + }) +} + func (d *saaDriver) start_Poll_FailRetryably_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores - d.startRetryable(t, backingOffInterval) // long, so the pause and read land well before the retry dispatches + d.startRetryable(t, backingOffInterval, 3) // long, so the pause and read land well before the retry dispatches d.pollTask(t) d.failRetryably(t) d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) @@ -276,7 +367,7 @@ func (d *saaDriver) start_Poll_FailRetryably_Paused(t *testing.T) activityInfoPr } func (d *saaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores - d.startRetryable(t, dispatchInterval) + d.startRetryable(t, dispatchInterval, 3) d.pollTask(t) d.failRetryably(t) d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) @@ -287,10 +378,10 @@ func (d *saaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) }) } -// startRetryable starts an activity whose failures are retryable, with a constant backoff of retryInterval. -func (d *saaDriver) startRetryable(t *testing.T, retryInterval time.Duration) { - id := testcore.RandomizeStr(t.Name()) - resp, err := d.env.FrontendClient().StartActivityExecution(d.s.Context(), &workflowservice.StartActivityExecutionRequest{ +// 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(), @@ -302,10 +393,25 @@ func (d *saaDriver) startRetryable(t *testing.T, retryInterval time.Duration) { InitialInterval: durationpb.New(retryInterval), BackoffCoefficient: 1.0, MaximumInterval: durationpb.New(retryInterval), - MaximumAttempts: 3, + 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 } @@ -317,6 +423,14 @@ func (d *saaDriver) failRetryably(t *testing.T) { 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(), @@ -443,8 +557,16 @@ 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) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_STARTED && p.Attempt == 1 + }) +} + func (d *wfaDriver) start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores - d.startRetryable(t, backingOffInterval) + d.startRetryable(t, backingOffInterval, 3) d.pollTask(t) d.failRetryably(t) return d.awaitObserve(t, func(p activityInfoProjection) bool { @@ -452,8 +574,17 @@ func (d *wfaDriver) start_Poll_FailRetryably_ObserveBackingOff(t *testing.T) act }) } +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) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED && p.Attempt == 2 + }) +} + func (d *wfaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores - d.startRetryable(t, dispatchInterval) + d.startRetryable(t, dispatchInterval, 3) d.pollTask(t) d.failRetryably(t) d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) @@ -461,8 +592,20 @@ func (d *wfaDriver) start_Poll_FailRetryably_RetryDispatched(t *testing.T) activ return d.observe(t) } +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.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) + time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching + d.pollTask(t) + return d.awaitObserve(t, func(p activityInfoProjection) bool { + return p.State == enumspb.PENDING_ACTIVITY_STATE_STARTED && p.Attempt == 2 + }) +} + func (d *wfaDriver) start_Poll_FailRetryably_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores - d.startRetryable(t, backingOffInterval) // long, so the pause and read land well before the retry dispatches + d.startRetryable(t, backingOffInterval, 3) // long, so the pause and read land well before the retry dispatches d.pollTask(t) d.failRetryably(t) d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) @@ -473,7 +616,7 @@ func (d *wfaDriver) start_Poll_FailRetryably_Paused(t *testing.T) activityInfoPr } func (d *wfaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores - d.startRetryable(t, dispatchInterval) + d.startRetryable(t, dispatchInterval, 3) d.pollTask(t) d.failRetryably(t) d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) @@ -486,7 +629,7 @@ func (d *wfaDriver) start_Poll_FailRetryably_BackoffElapses_Paused(t *testing.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) { +func (d *wfaDriver) startRetryable(t *testing.T, retryInterval time.Duration, maxAttempts int32) { wfTQ := testcore.RandomizeStr("parity-wf") d.activityTQ = testcore.RandomizeStr("parity-act") @@ -498,7 +641,7 @@ func (d *wfaDriver) startRetryable(t *testing.T, retryInterval time.Duration) { 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: 3, + TaskQueue: d.activityTQ, StartToClose: time.Hour, RetryInterval: retryInterval, MaxAttempts: maxAttempts, }) require.NoError(t, err) d.run = run @@ -511,6 +654,14 @@ func (d *wfaDriver) failRetryably(t *testing.T) { 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, @@ -593,3 +744,14 @@ func retryableFailure() *failurepb.Failure { }, } } + +// 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") - }) -} From 5676d92ad4ce1885293dc0662f89ba1e40925f56 Mon Sep 17 00:00:00 2001 From: Dan Davison Date: Thu, 23 Jul 2026 16:39:41 -0400 Subject: [PATCH 8/9] Replace fixed sleeps with long-polls / awaits in the retry parity drivers The drivers waited out the retry backoff with time.Sleep. Instead: where the next step is a poll, issue the long-poll directly (PollActivityTaskQueue blocks until the retry dispatches); where we observe the dispatched-but-unpolled state, awaitObserve until the projection shows the dispatch (next-schedule cleared). Tracks the real transition rather than a fixed wait; the subtest wall-clock drops ~4x. Run: go test -tags test_dep -run 'TestStandaloneActivityTestSuite/TestParityCurrentRetryInterval' -count=1 -v -timeout 580s ./tests/ Observe running state with a single describe, not a client-side poll loop PollActivityTaskQueue returns only after Matching records the attempt's start on the History shard, and DescribeActivityExecution reads that shard strongly-consistently, so the STARTED state is already visible when the poll returns. The observe-running cases now do a single describe instead of awaitObserve. Verified deterministic over repeated runs. Run: go test -tags test_dep -run 'TestStandaloneActivityTestSuite/TestParityCurrentRetryInterval' -count=1 -v -timeout 580s ./tests/ Use single describe after synchronous RPCs; reserve polling for dispatch-delay elapses Apply the driver waiting taxonomy (see dandavison/log#272): a synchronous RPC (RespondActivityTaskFailed, PauseActivityExecution) commits its transition on the History shard before returning, and Describe reads that shard strongly-consistently, so the backing-off / next-retry-delay / paused-before-dispatch cases observe with a single describe instead of a client-poll loop. awaitObserve is now reserved for the dispatch-delay elapse (backoff), whose read-time projection flip bumps no version for a long-poll to wake on. Verified over repeated runs. Run: go test -tags test_dep -run 'TestStandaloneActivityTestSuite/TestParityCurrentRetryInterval' -count=1 -v -timeout 580s ./tests/ --- tests/activity_parity_test.go | 116 ++++++++++++++-------------------- 1 file changed, 46 insertions(+), 70 deletions(-) diff --git a/tests/activity_parity_test.go b/tests/activity_parity_test.go index 5e544c6ce2c..379dfc620fa 100644 --- a/tests/activity_parity_test.go +++ b/tests/activity_parity_test.go @@ -199,6 +199,12 @@ type activityInfoProjection struct { 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 @@ -310,72 +316,54 @@ func (d *saaDriver) describeActivity(t *testing.T) *workflowservice.DescribeActi func (d *saaDriver) start_Poll_ObserveRunning(t *testing.T) activityInfoProjection { //nolint:staticcheck // ST1003: underscores d.startRetryable(t, backingOffInterval, 3) - d.pollTask(t) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_STARTED && p.Attempt == 1 - }) + 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) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED && p.Attempt == 2 - }) + 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) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED && p.Attempt == 2 - }) + 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) - d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) - time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching - return d.observe(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.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) - time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching - d.pollTask(t) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_STARTED && p.Attempt == 2 - }) + 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 and read land well before the retry dispatches + d.startRetryable(t, backingOffInterval, 3) // long, so the pause lands well before the retry dispatches d.pollTask(t) - d.failRetryably(t) - d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) - d.pauseActivity(t) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_PAUSED - }) + 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, func(p activityInfoProjection) bool { return p.Attempt == 2 }) - time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching - d.pauseActivity(t) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_PAUSED - }) + 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 @@ -449,7 +437,10 @@ func (d *saaDriver) observe(t *testing.T) activityInfoProjection { } } -// awaitObserve polls observe until pred holds, returning that projection. +// awaitObserve client-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 { deadline := time.Now().Add(15 * time.Second) for { @@ -559,72 +550,54 @@ 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) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_STARTED && p.Attempt == 1 - }) + 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) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED && p.Attempt == 2 - }) + 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) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED && p.Attempt == 2 - }) + 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) - d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) - time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching - return d.observe(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.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) - time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching - d.pollTask(t) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_STARTED && p.Attempt == 2 - }) + 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 and read land well before the retry dispatches + d.startRetryable(t, backingOffInterval, 3) // long, so the pause lands well before the retry dispatches d.pollTask(t) - d.failRetryably(t) - d.awaitObserve(t, func(p activityInfoProjection) bool { return p.Attempt == 2 }) - d.pauseActivity(t) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_PAUSED - }) + 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, func(p activityInfoProjection) bool { return p.Attempt == 2 }) - time.Sleep(dispatchInterval + 3*time.Second) // let the backoff elapse so the retry dispatches to Matching - d.pauseActivity(t) - return d.awaitObserve(t, func(p activityInfoProjection) bool { - return p.State == enumspb.PENDING_ACTIVITY_STATE_PAUSED - }) + 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 @@ -688,7 +661,10 @@ func (d *wfaDriver) observe(t *testing.T) activityInfoProjection { return activityInfoProjection{} } -// awaitObserve polls observe until pred holds, returning that projection. +// awaitObserve client-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 { deadline := time.Now().Add(15 * time.Second) for { From c1ac44a964301f17fb416c4eeef911413ecd870e Mon Sep 17 00:00:00 2001 From: Dan Davison Date: Thu, 23 Jul 2026 17:59:51 -0400 Subject: [PATCH 9/9] Poll via await.Require instead of a hand-rolled time.Sleep loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dispatch-delay projection poll used a hand-rolled loop with time.Sleep (forbidden by forbidigo). Switch awaitObserve to await.Require (the blessed polling helper): observe/describeActivity now take require.TestingT so the poll's describe runs against the callback's *await.T, making transient RPC errors retryable rather than a hard failure. Plain require.Eventually doesn't fit — it runs the condition in a goroutine (so observe's require calls would Goexit it) and returns only a bool, not the projection. Run: go test -tags test_dep -run 'TestStandaloneActivityTestSuite/TestParityCurrentRetryInterval' -count=1 -v -timeout 580s ./tests/ --- tests/activity_parity_test.go | 65 +++++++++++++++-------------------- 1 file changed, 28 insertions(+), 37 deletions(-) diff --git a/tests/activity_parity_test.go b/tests/activity_parity_test.go index 379dfc620fa..822d39039ea 100644 --- a/tests/activity_parity_test.go +++ b/tests/activity_parity_test.go @@ -18,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" ) @@ -303,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, @@ -427,7 +429,7 @@ func (d *saaDriver) pauseActivity(t *testing.T) { } // observe reads the activity's public retry-scheduling info as the shared projection. -func (d *saaDriver) observe(t *testing.T) activityInfoProjection { +func (d *saaDriver) observe(t require.TestingT) activityInfoProjection { i := d.describeActivity(t).GetInfo() return activityInfoProjection{ State: i.GetRunState(), @@ -437,23 +439,17 @@ func (d *saaDriver) observe(t *testing.T) activityInfoProjection { } } -// awaitObserve client-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. +// 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 { - deadline := time.Now().Add(15 * time.Second) - for { - p := d.observe(t) - if pred(p) { - return p - } - if time.Now().After(deadline) { - require.Fail(t, "activity did not reach the expected state", "last observed: %+v", p) - return p - } - time.Sleep(100 * time.Millisecond) - } + 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 -------------------------------------------------------------- @@ -643,8 +639,9 @@ func (d *wfaDriver) pauseActivity(t *testing.T) { require.NoError(t, err) } -// observe reads the activity's public retry-scheduling info via DescribeWorkflowExecution, as the shared projection. -func (d *wfaDriver) observe(t *testing.T) activityInfoProjection { +// 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() { @@ -661,23 +658,17 @@ func (d *wfaDriver) observe(t *testing.T) activityInfoProjection { return activityInfoProjection{} } -// awaitObserve client-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. +// 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 { - deadline := time.Now().Add(15 * time.Second) - for { - p := d.observe(t) - if pred(p) { - return p - } - if time.Now().After(deadline) { - require.Fail(t, "activity did not reach the expected state", "last observed: %+v", p) - return p - } - time.Sleep(100 * time.Millisecond) - } + 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.