@@ -5900,6 +5900,150 @@ func (s *Versioning3Suite) TestRemoveOverride_ClearsDeclinedState() {
59005900 })
59015901}
59025902
5903+ // TestStalePartition_RevisionSuppressesTrampolining verifies that when a stale
5904+ // matching partition sends an outdated target version, the revision-based
5905+ // comparison in case 4 suppresses the signal and prevents trampolining.
5906+ //
5907+ // Flow:
5908+ // 1. Start pinned workflow on v1, set v1 as current
5909+ // 2. Set v2 as current (revision increments)
5910+ // 3. Set v3 as current (revision increments again)
5911+ // 4. Trigger WFT → signal fires (target=v3 at high revision), CaN with decline
5912+ // 5. Roll back task queue to v2 with revision 0 (simulating stale partition)
5913+ // 6. Trigger WFT → assert targetDeploymentVersionChanged=false (revision 0 <= declined revision)
5914+ // 7. Set v4 as current (fresh version with higher revision)
5915+ // 8. Trigger WFT → assert targetDeploymentVersionChanged=true (new revision > declined revision)
5916+ func (s * Versioning3Suite ) TestStalePartition_RevisionSuppressesTrampolining () {
5917+ s .OverrideDynamicConfig (dynamicconfig .MatchingNumTaskqueueReadPartitions , 1 )
5918+ s .OverrideDynamicConfig (dynamicconfig .MatchingNumTaskqueueWritePartitions , 1 )
5919+
5920+ ctx , cancel := context .WithTimeout (context .Background (), 60 * time .Second )
5921+ defer cancel ()
5922+
5923+ tv1 := testvars .New (s ).WithBuildIDNumber (1 )
5924+ tv2 := tv1 .WithBuildIDNumber (2 )
5925+ tv3 := tv1 .WithBuildIDNumber (3 )
5926+
5927+ // Start async poller for v1 that handles first WFT and declares pinned behavior
5928+ wftCompleted := make (chan struct {})
5929+ s .pollWftAndHandle (tv1 , false , wftCompleted ,
5930+ func (task * workflowservice.PollWorkflowTaskQueueResponse ) (* workflowservice.RespondWorkflowTaskCompletedRequest , error ) {
5931+ s .NotNil (task )
5932+ return respondEmptyWft (tv1 , false , vbPinned ), nil
5933+ })
5934+
5935+ // Register v1 and set it as current
5936+ s .waitForDeploymentDataPropagation (tv1 , versionStatusInactive , false , tqTypeWf )
5937+ s .setCurrentDeployment (tv1 )
5938+
5939+ // Start workflow — first WFT handled by async poller (pinned on v1)
5940+ runID := s .startWorkflow (tv1 , nil )
5941+ execution := tv1 .WithRunID (runID ).WorkflowExecution ()
5942+ s .WaitForChannel (ctx , wftCompleted )
5943+ s .verifyWorkflowVersioning (s .Assertions , tv1 , vbPinned , tv1 .Deployment (), nil , nil )
5944+
5945+ // Register v2, set v2 as current (revision increments)
5946+ s .idlePollWorkflow (ctx , tv2 , true , ver3MinPollTime , "v2 poller registration" )
5947+ s .setCurrentDeployment (tv2 )
5948+
5949+ // Register v3, set v3 as current (revision increments again)
5950+ s .idlePollWorkflow (ctx , tv3 , true , ver3MinPollTime , "v3 poller registration" )
5951+ s .setCurrentDeployment (tv3 )
5952+
5953+ // Trigger WFT — target should be v3 with a high revision
5954+ s .triggerNormalWFT (ctx , tv1 , execution )
5955+
5956+ // Process: targetDeploymentVersionChanged=true → CaN without AU (decline v3)
5957+ s .pollWftAndHandle (tv1 , false , nil ,
5958+ func (task * workflowservice.PollWorkflowTaskQueueResponse ) (* workflowservice.RespondWorkflowTaskCompletedRequest , error ) {
5959+ s .NotNil (task )
5960+ var lastStarted * historypb.HistoryEvent
5961+ for _ , event := range task .History .GetEvents () {
5962+ if event .GetEventType () == enumspb .EVENT_TYPE_WORKFLOW_TASK_STARTED {
5963+ lastStarted = event
5964+ }
5965+ }
5966+ s .NotNil (lastStarted )
5967+ s .True (lastStarted .GetWorkflowTaskStartedEventAttributes ().GetTargetWorkerDeploymentVersionChanged (),
5968+ "expected true after v3 becomes current" )
5969+
5970+ return & workflowservice.RespondWorkflowTaskCompletedRequest {
5971+ Commands : []* commandpb.Command {
5972+ {
5973+ CommandType : enumspb .COMMAND_TYPE_CONTINUE_AS_NEW_WORKFLOW_EXECUTION ,
5974+ Attributes : & commandpb.Command_ContinueAsNewWorkflowExecutionCommandAttributes {
5975+ ContinueAsNewWorkflowExecutionCommandAttributes : & commandpb.ContinueAsNewWorkflowExecutionCommandAttributes {
5976+ WorkflowType : tv1 .WorkflowType (),
5977+ TaskQueue : tv1 .TaskQueue (),
5978+ Input : tv1 .Any ().Payloads (),
5979+ },
5980+ },
5981+ },
5982+ },
5983+ VersioningBehavior : vbPinned ,
5984+ DeploymentOptions : tv1 .WorkerDeploymentOptions (true ),
5985+ }, nil
5986+ })
5987+
5988+ // CaN run: first WFT — declined=v3 propagated from previous run
5989+ s .pollWftAndHandle (tv1 , false , nil ,
5990+ func (task * workflowservice.PollWorkflowTaskQueueResponse ) (* workflowservice.RespondWorkflowTaskCompletedRequest , error ) {
5991+ s .NotNil (task )
5992+ s .NotEqual (execution .RunId , task .WorkflowExecution .RunId ,
5993+ "CaN should have created a new run" )
5994+ execution = task .WorkflowExecution
5995+ return respondEmptyWft (tv1 , false , vbPinned ), nil
5996+ })
5997+
5998+ // Simulate stale partition: roll back task queue to v2 with revision 0
5999+ s .rollbackTaskQueueToVersion (tv2 )
6000+
6001+ // Trigger WFT with stale data — target is now v2 at revision 0
6002+ s .triggerNormalWFT (ctx , tv1 , execution )
6003+
6004+ // Assert: revision 0 <= declined revision → signal suppressed
6005+ s .pollWftAndHandle (tv1 , false , nil ,
6006+ func (task * workflowservice.PollWorkflowTaskQueueResponse ) (* workflowservice.RespondWorkflowTaskCompletedRequest , error ) {
6007+ s .NotNil (task )
6008+ var lastStarted * historypb.HistoryEvent
6009+ for _ , event := range task .History .GetEvents () {
6010+ if event .GetEventType () == enumspb .EVENT_TYPE_WORKFLOW_TASK_STARTED {
6011+ lastStarted = event
6012+ }
6013+ }
6014+ s .NotNil (lastStarted )
6015+ s .False (lastStarted .GetWorkflowTaskStartedEventAttributes ().GetTargetWorkerDeploymentVersionChanged (),
6016+ "stale partition (revision 0) should be suppressed by declined revision" )
6017+ return respondEmptyWft (tv1 , false , vbPinned ), nil
6018+ })
6019+
6020+ // Set a new v4 as current — this produces a revision strictly higher than
6021+ // the declined revision, simulating an up-to-date partition with fresh data.
6022+ tv4 := tv1 .WithBuildIDNumber (4 )
6023+ s .idlePollWorkflow (ctx , tv4 , true , ver3MinPollTime , "v4 poller registration" )
6024+ s .setCurrentDeployment (tv4 )
6025+ s .waitForDeploymentDataPropagation (tv4 , versionStatusCurrent , false , tqTypeWf )
6026+
6027+ // Trigger WFT with fresh data — target is v4 at higher revision
6028+ s .triggerNormalWFT (ctx , tv1 , execution )
6029+
6030+ // Assert: new revision > declined revision → signal fires
6031+ s .pollWftAndHandle (tv1 , false , nil ,
6032+ func (task * workflowservice.PollWorkflowTaskQueueResponse ) (* workflowservice.RespondWorkflowTaskCompletedRequest , error ) {
6033+ s .NotNil (task )
6034+ var lastStarted * historypb.HistoryEvent
6035+ for _ , event := range task .History .GetEvents () {
6036+ if event .GetEventType () == enumspb .EVENT_TYPE_WORKFLOW_TASK_STARTED {
6037+ lastStarted = event
6038+ }
6039+ }
6040+ s .NotNil (lastStarted )
6041+ s .True (lastStarted .GetWorkflowTaskStartedEventAttributes ().GetTargetWorkerDeploymentVersionChanged (),
6042+ "up-to-date partition with higher revision should fire signal" )
6043+ return respondCompleteWorkflow (tv1 , vbPinned ), nil
6044+ })
6045+ }
6046+
59036047// TestRetryOfDeclinedCaN_SignalsOnNewTarget verifies that when a CaN'd run
59046048// ,which declined to upgrade, fails and is retried by the server, the retry
59056049// run inherits NotificationSuppressedTargetVersion from the original CaN
0 commit comments