Merge branch 'master' into upgrade-dd-trace-go-to-v2

Author: yarinzirlin

contrib/tools/workflowcheck/README.md

@@ -145,7 +145,7 @@ In addition to those functions/vars, the following Go source constructs are cons
 * Iterating over a channel via `range`
 * Iterating over a map via `range`
 
-Many constructs that are known to be non-deterministic, such as mutating a global variable, are not able to be reliably
+Many constructs that are known to be non-deterministic, such as mutating a global variable, cannot be reliably
 distinguished from deterministic use in common cases. This tool intentionally does not flag them.
 
 In some cases, functions that are considered non-deterministic are commonly used in ways that only follow a

internal/internal_worker.go

@@ -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
+	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{}
@@ -518,7 +529,6 @@ func newActivityWorker(
 	} else {
 		slotSupplier = params.Tuner.GetActivityTaskSlotSupplier()
 	}
-
 	bwo := baseWorkerOptions{
 		pollerRate:       defaultPollerRate,
 		slotSupplier:     slotSupplier,

internal/version.go

@@ -8,7 +8,7 @@ const (
 	// Server validates if SDKVersion fits its supported range and rejects request if it doesn't.
 	//
 	// Exposed as: [go.temporal.io/sdk/temporal.SDKVersion]
-	SDKVersion = "1.37.0"
+	SDKVersion = "1.38.0"
 
 	// SDKName represents the name of the SDK.
 	SDKName = clientNameHeaderValue

temporalnexus/operation.go

@@ -219,6 +219,9 @@ type WorkflowHandle[T any] interface {
 	// Link to the WorkflowExecutionStarted event of the workflow represented by this handle.
 	link() nexus.Link
 	token() string // Cached operation token
+
+	// typeMarker is a no-op method to associate the generic type T with the interface.
+	typeMarker(T)
 }
 
 type workflowHandle[T any] struct {
@@ -259,6 +262,8 @@ func (h workflowHandle[T]) token() string {
 	return h.cachedToken
 }
 
+func (h workflowHandle[T]) typeMarker(T) {}
+
 // ExecuteWorkflow starts a workflow run for a [WorkflowRunOperationOptions] Handler, linking the execution chain to a
 // Nexus operation (subsequent runs started from continue-as-new and retries).
 // Automatically propagates the callback and request ID from the nexus options to the workflow.

test/worker_deployment_test.go

@@ -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"
@@ -236,6 +237,60 @@ func (ts *WorkerDeploymentTestSuite) TestBuildIDChangesOverWorkflowLifetime() {
 	ts.Equal("2.0", lastBuildID)
 }
 
+func (ts *WorkerDeploymentTestSuite) TestBuildIDWithSession() {
+
+	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))
+}
+
 func (ts *WorkerDeploymentTestSuite) TestPinnedBehaviorThreeWorkers() {
 	if os.Getenv("DISABLE_SERVER_1_27_TESTS") != "" {
 		ts.T().Skip("temporal server 1.27+ required")