Merge branch 'uber-go:master' into master

.github/workflows/breaking_change_pr_template.md

@@ -0,0 +1,18 @@
+**Detailed Description**
+[In-depth description of the changes made to the interfaces, specifying new fields, removed fields, or modified data structures]
+
+**Impact Analysis**
+- **Backward Compatibility**: [Analysis of backward compatibility]
+- **Forward Compatibility**: [Analysis of forward compatibility]
+
+**Testing Plan**
+- **Unit Tests**: [Do we have unit test covering the change?]
+- **Persistence Tests**: [If the change is related to a data type which is persisted, do we have persistence tests covering the change?]
+- **Integration Tests**: [Do we have integration test covering the change?]
+- **Compatibility Tests**: [Have we done tests to test the backward and forward compatibility?]
+
+**Rollout Plan**
+- What is the rollout plan?
+- Does the order of deployment matter?
+- Is it safe to rollback? Does the order of rollback matter?
+- Is there a kill switch to mitigate the impact immediately?

.github/workflows/breaking_change_reminder.yml

@@ -0,0 +1,54 @@
+name: Workflow for Breaking Change Reminder
+on:
+  pull_request:
+    paths:
+      # below files do not cover all the exposed types/funcs, but it's a good start to detect potentially breaking changes
+      - activity/activity.go
+      - client/client.go
+      - encoded/encoded.go
+      - interceptors/workflow_interceptor.go
+      - internal/activity.go
+      - internal/client.go
+      - internal/encoded.go
+      - internal/workflow.go
+      - internal/interceptors.go
+      - internal/worker.go
+      - internal/workflow.go
+      - worker/worker.go
+      - workflow/*.go
+
+jobs:
+  breaking-change-pr-template-reminder:
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout
+      uses: actions/checkout@v4
+      with:
+        fetch-depth: 0
+
+    - name: Fail if PR description is missing breaking change template
+      if: steps.pr-changes.outputs.changes != '[]'
+      run: |
+        PR_NUMBER=${{ github.event.pull_request.number }}
+        PR_URL="https://api.github.com/repos/${{ github.repository }}/pulls/${PR_NUMBER}"
+        BODY=$(curl $PR_URL | jq '.body')
+        CHECKLIST=(
+          "Detailed Description"
+          "Impact Analysis"
+          "Testing Plan"
+          "Rollout Plan"
+        )
+        TEMPLATE=$(cat .github/workflows/breaking_change_pr_template.md)
+
+        for i in "${CHECKLIST[@]}"; do
+            if [[ "$BODY" == *"$i"* ]]; then
+                continue
+            else
+                echo "Potential breaking changes detected! Please update the PR description to include following template:"
+                echo "---"
+                echo "$TEMPLATE"
+                echo "---"
+                exit 1
+            fi
+        done

internal/internal_logging_tags.go

@@ -41,6 +41,7 @@ const (
 	tagVisibilityQuery             = "VisibilityQuery"
 	tagPanicError                  = "PanicError"
 	tagPanicStack                  = "PanicStack"
+	tagPollerType                  = "PollerType"
 	causeTag                       = "pollerrorcause"
 	tagWorkflowRuntimeLength       = "workflowruntimelength"
 	tagNonDeterminismDetectionType = "NonDeterminismDetectionType"

internal/internal_task_pollers.go

@@ -34,6 +34,7 @@ import (
 	"github.com/pborman/uuid"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	s "go.uber.org/cadence/.gen/go/shared"
@@ -248,30 +249,28 @@ func (bp *basePoller) doPoll(
 	var err error
 	var result interface{}
 
-	doneC := make(chan struct{})
 	ctx, cancel, _ := newChannelContext(context.Background(), featureFlags, chanTimeout(pollTaskServiceTimeOut))
+	defer cancel()
 
 	go func() {
-		defer close(doneC)
+		defer cancel()
 		defer func() {
 			if p := recover(); p != nil {
 				bp.metricsScope.Counter(metrics.InternalPanicCounter).Inc(1)
 				st := getStackTraceRaw("base poller [panic]:", 7, 0)
-				bp.logger.Error("Unhandled panic.",
+				bp.logger.Error("Unhandled panic",
 					zap.String(tagPanicError, fmt.Sprintf("%v", p)),
 					zap.String(tagPanicStack, st))
 				err = newPanicError(p, st)
 			}
 		}()
-		defer cancel()
 		result, err = pollFunc(ctx)
 	}()
 
 	select {
-	case <-doneC:
+	case <-ctx.Done():
 		return result, err
 	case <-bp.shutdownC:
-		cancel()
 		return nil, errShutdown
 	}
 }
@@ -288,7 +287,7 @@ func newWorkflowTaskPoller(
 		basePoller: basePoller{
 			shutdownC:    params.WorkerStopChannel,
 			metricsScope: metrics.NewTaggedScope(params.MetricsScope),
-			logger:       params.Logger,
+			logger:       params.Logger.With(zapcore.Field{Key: tagPollerType, Type: zapcore.StringType, String: "Workflow"}),
 		},
 		service:                      service,
 		domain:                       domain,
@@ -542,7 +541,7 @@ func newLocalActivityPoller(params workerExecutionParameters, laTunnel *localAct
 		basePoller: basePoller{
 			shutdownC:    params.WorkerStopChannel,
 			metricsScope: metrics.NewTaggedScope(params.MetricsScope),
-			logger:       params.Logger,
+			logger:       params.Logger.With(zapcore.Field{Key: tagPollerType, Type: zapcore.StringType, String: "LocalActivity"}),
 		},
 		handler:  handler,
 		laTunnel: laTunnel,
@@ -1002,7 +1001,7 @@ func newActivityTaskPoller(taskHandler ActivityTaskHandler, service workflowserv
 	activityTaskPoller := &activityTaskPoller{
 		basePoller: basePoller{
 			shutdownC:    params.WorkerStopChannel,
-			logger:       params.Logger,
+			logger:       params.Logger.With(zap.Field{Key: tagPollerType, Type: zapcore.StringType, String: "Activity"}),
 			metricsScope: metrics.NewTaggedScope(params.MetricsScope),
 		},
 		taskHandler:         taskHandler,

internal/internal_task_pollers_test.go

@@ -85,6 +85,33 @@ func TestActivityTaskPollerHandlesPanics(t *testing.T) {
 	assert.Equal(t, "oh no", panicErr.value)
 }
 
+func TestActivityTaskPollerHandlesCancel(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	service := workflowservicetest.NewMockClient(ctrl)
+	workerStopChannel := make(chan struct{}, 1)
+	pollBlockingChannel := make(chan struct{}, 1)
+	defer close(pollBlockingChannel)
+	activityPoller := newActivityTaskPoller(nil, service, "test", workerExecutionParameters{
+		TaskList: "tasklist",
+
+		WorkerStopChannel: workerStopChannel,
+		WorkerOptions: WorkerOptions{
+			Identity: "identity",
+			Logger:   zaptest.NewLogger(t),
+		},
+	})
+	service.EXPECT().PollForActivityTask(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, _ *shared.PollForActivityTaskRequest, opts ...yarpc.CallOption) (*shared.PollForActivityTaskResponse, error) {
+		workerStopChannel <- struct{}{}
+		<-pollBlockingChannel
+		return nil, nil
+	})
+
+	result, err := activityPoller.PollTask()
+
+	assert.Nil(t, result)
+	assert.ErrorIs(t, err, errShutdown)
+}
+
 func TestWorkflowTaskPollerHandlesPanics(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	service := workflowservicetest.NewMockClient(ctrl)