Skip to content

Stop sending deployment info for session per-worker task queue #2088

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
+67 −2
Open

Stop sending deployment info for session per-worker task queue #2088

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
95e17e0
Stop sending deployment info for session per-worker task queue
Quinn-With-Two-Ns Oct 22, 2025
8e80d65
Remove skip
Quinn-With-Two-Ns Oct 23, 2025
2580e74
Stop worker
Quinn-With-Two-Ns Oct 23, 2025
5d9c0b3
Fix test
Quinn-With-Two-Ns Oct 24, 2025
f057f32
Set UseBuildIDForVersioning to false
Quinn-With-Two-Ns Oct 28, 2025
5fc9d01
Just disable UseVersioning but still still send deployment name for d…
Quinn-With-Two-Ns Nov 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions internal/internal_worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -448,15 +448,26 @@ func newSessionWorker(client *WorkflowClient, params workerExecutionParameters,
creationTaskqueue := getCreationTaskqueue(params.TaskQueue)
params.BackgroundContext = context.WithValue(params.BackgroundContext, sessionEnvironmentContextKey, sessionEnvironment)
params.TaskQueue = sessionEnvironment.GetResourceSpecificTaskqueue()
// For the resource specific task queue, we don't need to include deployment options
// Save them to restore later
deployments := params.DeploymentOptions
Copy link
Contributor

@ShahabT ShahabT Oct 28, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Only disable versioning instead of removing the whole thing? better to keep the deployment name and build id if user has given it (to keep in the events).

Copy link
Contributor Author

@Quinn-With-Two-Ns Quinn-With-Two-Ns Oct 28, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does it make any difference for an activity only task queue? I don't see any difference in the history.

Copy link
Contributor

@ShahabT ShahabT Oct 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we might add the deployment options in activity events later

useBuildIDForVersioning := params.UseBuildIDForVersioning
// Disable versioning for activity worker within session, but still send deployment name for debug purpose
params.DeploymentOptions.UseVersioning = false
params.UseBuildIDForVersioning = false
activityWorker := newActivityWorker(client, params,
&workerOverrides{slotSupplier: params.Tuner.GetSessionActivitySlotSupplier()}, env, nil)
&workerOverrides{
slotSupplier: params.Tuner.GetSessionActivitySlotSupplier(),
}, env, nil)

params.ActivityTaskPollerBehavior = NewPollerBehaviorSimpleMaximum(
PollerBehaviorSimpleMaximumOptions{
MaximumNumberOfPollers: 1,
},
)
params.TaskQueue = creationTaskqueue
params.DeploymentOptions = deployments
params.UseBuildIDForVersioning = useBuildIDForVersioning
// Although we have session token bucket to limit session size across creation
// and recreation, we also limit it here for creation only
overrides := &workerOverrides{}
Expand Down Expand Up @@ -518,7 +529,6 @@ func newActivityWorker(
} else {
slotSupplier = params.Tuner.GetActivityTaskSlotSupplier()
}

bwo := baseWorkerOptions{
pollerRate: defaultPollerRate,
slotSupplier: slotSupplier,
Expand Down
55 changes: 55 additions & 0 deletions test/worker_deployment_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import (
enumspb "go.temporal.io/api/enums/v1"
"go.temporal.io/api/workflowservice/v1"

"go.temporal.io/sdk/activity"
"go.temporal.io/sdk/client"
"go.temporal.io/sdk/worker"
"go.temporal.io/sdk/workflow"
Expand Down Expand Up @@ -236,6 +237,60 @@ func (ts *WorkerDeploymentTestSuite) TestBuildIDChangesOverWorkflowLifetime() {
ts.Equal("2.0", lastBuildID)
}

func (ts *WorkerDeploymentTestSuite) TestBuildIDWithSession() {

Copy link
Contributor

@yuandrew yuandrew Nov 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we want to gate this? All other tests here have this req

if os.Getenv("DISABLE_SERVER_1_27_TESTS") != "" {
		ts.T().Skip("temporal server 1.27+ required")
	}

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

deploymentName := "deploy-test-" + uuid.NewString()
v1 := worker.WorkerDeploymentVersion{
DeploymentName: deploymentName,
BuildID: "1.0",
}

worker := worker.New(ts.client, ts.taskQueueName, worker.Options{
EnableSessionWorker: true,
DeploymentOptions: worker.DeploymentOptions{
UseVersioning: true,
Version: v1,
},
})

worker.RegisterWorkflowWithOptions(ts.workflows.BasicSession, workflow.RegisterOptions{
Name: "SessionBuildIDWorkflow",
VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade,
})

activities2 := &Activities2{}
result := &Activities{activities2: activities2}
activities2.impl = result
worker.RegisterActivityWithOptions(activities2, activity.RegisterOptions{Name: "Prefix_", DisableAlreadyRegisteredCheck: true})

ts.NoError(worker.Start())
defer worker.Stop()

dHandle := ts.client.WorkerDeploymentClient().GetHandle(deploymentName)

ts.waitForWorkerDeployment(ctx, dHandle)

response1, err := dHandle.Describe(ctx, client.WorkerDeploymentDescribeOptions{})
ts.NoError(err)

ts.waitForWorkerDeploymentVersion(ctx, dHandle, v1)

_, err = dHandle.SetCurrentVersion(ctx, client.WorkerDeploymentSetCurrentVersionOptions{
BuildID: v1.BuildID,
ConflictToken: response1.ConflictToken,
})
ts.NoError(err)

// start workflow1 with 1.0, BasicSession, auto-upgrade
wfHandle, err := ts.client.ExecuteWorkflow(ctx, ts.startWorkflowOptions("evolving-wf-1"), "SessionBuildIDWorkflow")
ts.NoError(err)

ts.NoError(wfHandle.Get(ctx, nil))
Copy link
Contributor

@yuandrew yuandrew Nov 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is this actually verifying? This seems to pass on master without your fix

Copy link
Contributor

@yuandrew yuandrew Nov 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we be asserting that buildId and versioning settings for the session activity worker is properly set?

}
Copy link

@cursor cursor bot Oct 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: Test Missing Version Check for Worker Deployment

The new TestBuildIDWithSession test uses worker deployment versioning features, which require Temporal server 1.27+. It's missing the server version skip condition found in other deployment tests, potentially causing failures on older server versions.

Fix in Cursor Fix in Web


func (ts *WorkerDeploymentTestSuite) TestPinnedBehaviorThreeWorkers() {
if os.Getenv("DISABLE_SERVER_1_27_TESTS") != "" {
ts.T().Skip("temporal server 1.27+ required")
Expand Down
Loading