Revert "Handle panics while polling for tasks (#1352)"

internal/common/metrics/constants.go

@@ -85,7 +85,6 @@ const (
 	ActivityLocalDispatchFailedCounter          = CadenceMetricsPrefix + "activity-local-dispatch-failed"
 	ActivityLocalDispatchSucceedCounter         = CadenceMetricsPrefix + "activity-local-dispatch-succeed"
 	WorkerPanicCounter                          = CadenceMetricsPrefix + "worker-panic"
-	InternalPanicCounter                        = CadenceMetricsPrefix + "internal-panic"
 
 	UnhandledSignalsCounter = CadenceMetricsPrefix + "unhandled-signals"
 	CorruptedSignalsCounter = CadenceMetricsPrefix + "corrupted-signals"

internal/internal_logging_tags.go

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

internal/internal_task_pollers.go

@@ -34,7 +34,6 @@ 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"
@@ -64,9 +63,7 @@ type (
 
 	// basePoller is the base class for all poller implementations
 	basePoller struct {
-		shutdownC    <-chan struct{}
-		metricsScope *metrics.TaggedScope
-		logger       *zap.Logger
+		shutdownC <-chan struct{}
 	}
 
 	// workflowTaskPoller implements polling/processing a workflow task
@@ -78,6 +75,8 @@ type (
 		service      workflowserviceclient.Interface
 		taskHandler  WorkflowTaskHandler
 		ldaTunnel    *locallyDispatchedActivityTunnel
+		metricsScope *metrics.TaggedScope
+		logger       *zap.Logger
 
 		stickyUUID                   string
 		disableStickyExecution       bool
@@ -98,6 +97,8 @@ type (
 		identity            string
 		service             workflowserviceclient.Interface
 		taskHandler         ActivityTaskHandler
+		metricsScope        *metrics.TaggedScope
+		logger              *zap.Logger
 		activitiesPerSecond float64
 		featureFlags        FeatureFlags
 	}
@@ -122,8 +123,10 @@ type (
 
 	localActivityTaskPoller struct {
 		basePoller
-		handler  *localActivityTaskHandler
-		laTunnel *localActivityTunnel
+		handler      *localActivityTaskHandler
+		metricsScope tally.Scope
+		logger       *zap.Logger
+		laTunnel     *localActivityTunnel
 	}
 
 	localActivityTaskHandler struct {
@@ -249,28 +252,20 @@ 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 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",
-					zap.String(tagPanicError, fmt.Sprintf("%v", p)),
-					zap.String(tagPanicStack, st))
-				err = newPanicError(p, st)
-			}
-		}()
 		result, err = pollFunc(ctx)
+		cancel()
+		close(doneC)
 	}()
 
 	select {
-	case <-ctx.Done():
+	case <-doneC:
 		return result, err
 	case <-bp.shutdownC:
+		cancel()
 		return nil, errShutdown
 	}
 }
@@ -284,17 +279,15 @@ func newWorkflowTaskPoller(
 	params workerExecutionParameters,
 ) *workflowTaskPoller {
 	return &workflowTaskPoller{
-		basePoller: basePoller{
-			shutdownC:    params.WorkerStopChannel,
-			metricsScope: metrics.NewTaggedScope(params.MetricsScope),
-			logger:       params.Logger.With(zapcore.Field{Key: tagPollerType, Type: zapcore.StringType, String: "Workflow"}),
-		},
+		basePoller:                   basePoller{shutdownC: params.WorkerStopChannel},
 		service:                      service,
 		domain:                       domain,
 		taskListName:                 params.TaskList,
 		identity:                     params.Identity,
 		taskHandler:                  taskHandler,
 		ldaTunnel:                    ldaTunnel,
+		metricsScope:                 metrics.NewTaggedScope(params.MetricsScope),
+		logger:                       params.Logger,
 		stickyUUID:                   uuid.New(),
 		disableStickyExecution:       params.DisableStickyExecution,
 		StickyScheduleToStartTimeout: params.StickyScheduleToStartTimeout,
@@ -538,13 +531,11 @@ func newLocalActivityPoller(params workerExecutionParameters, laTunnel *localAct
 		tracer:             params.Tracer,
 	}
 	return &localActivityTaskPoller{
-		basePoller: basePoller{
-			shutdownC:    params.WorkerStopChannel,
-			metricsScope: metrics.NewTaggedScope(params.MetricsScope),
-			logger:       params.Logger.With(zapcore.Field{Key: tagPollerType, Type: zapcore.StringType, String: "LocalActivity"}),
-		},
-		handler:  handler,
-		laTunnel: laTunnel,
+		basePoller:   basePoller{shutdownC: params.WorkerStopChannel},
+		handler:      handler,
+		metricsScope: params.MetricsScope,
+		logger:       params.Logger,
+		laTunnel:     laTunnel,
 	}
 }
 
@@ -999,16 +990,14 @@ func newActivityTaskPoller(taskHandler ActivityTaskHandler, service workflowserv
 	domain string, params workerExecutionParameters) *activityTaskPoller {
 
 	activityTaskPoller := &activityTaskPoller{
-		basePoller: basePoller{
-			shutdownC:    params.WorkerStopChannel,
-			logger:       params.Logger.With(zap.Field{Key: tagPollerType, Type: zapcore.StringType, String: "Activity"}),
-			metricsScope: metrics.NewTaggedScope(params.MetricsScope),
-		},
+		basePoller:          basePoller{shutdownC: params.WorkerStopChannel},
 		taskHandler:         taskHandler,
 		service:             service,
 		domain:              domain,
 		taskListName:        params.TaskList,
 		identity:            params.Identity,
+		logger:              params.Logger,
+		metricsScope:        metrics.NewTaggedScope(params.MetricsScope),
 		activitiesPerSecond: params.TaskListActivitiesPerSecond,
 		featureFlags:        params.FeatureFlags,
 	}

internal/internal_task_pollers_test.go

@@ -26,14 +26,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/yarpc"
 	"go.uber.org/zap/zaptest"
-
-	"go.uber.org/cadence/.gen/go/cadence/workflowservicetest"
-	"go.uber.org/cadence/.gen/go/shared"
 )
 
 func TestLocalActivityPanic(t *testing.T) {
@@ -59,80 +54,3 @@ func TestLocalActivityPanic(t *testing.T) {
 	assert.Contains(t, perr.StackTrace(), "panic")
 	assert.Contains(t, perr.StackTrace(), t.Name(), "should mention the source location of the local activity that panicked")
 }
-
-func TestActivityTaskPollerHandlesPanics(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	service := workflowservicetest.NewMockClient(ctrl)
-	service.EXPECT().PollForActivityTask(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, _ *shared.PollForActivityTaskRequest, opts ...yarpc.CallOption) (*shared.PollForActivityTaskResponse, error) {
-		panic("oh no")
-	})
-	workerStopChannel := make(chan struct{}, 1)
-	activityPoller := newActivityTaskPoller(nil, service, "test", workerExecutionParameters{
-		TaskList: "tasklist",
-
-		WorkerStopChannel: workerStopChannel,
-		WorkerOptions: WorkerOptions{
-			Identity: "identity",
-			Logger:   zaptest.NewLogger(t),
-		},
-	})
-
-	result, err := activityPoller.PollTask()
-
-	assert.Nil(t, result)
-	var panicErr *PanicError
-	assert.ErrorAs(t, err, &panicErr)
-	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)
-	service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, _ *shared.PollForDecisionTaskRequest, opts ...yarpc.CallOption) (*shared.PollForDecisionTaskResponse, error) {
-		panic("oh no")
-	})
-	workerStopChannel := make(chan struct{}, 1)
-	workflowTaskPoller := newWorkflowTaskPoller(nil, nil, service, "test", workerExecutionParameters{
-		TaskList: "tasklist",
-
-		WorkerStopChannel: workerStopChannel,
-		WorkerOptions: WorkerOptions{
-			Identity: "identity",
-			Logger:   zaptest.NewLogger(t),
-		},
-	})
-
-	result, err := workflowTaskPoller.PollTask()
-
-	assert.Nil(t, result)
-	var panicErr *PanicError
-	assert.ErrorAs(t, err, &panicErr)
-	assert.Equal(t, "oh no", panicErr.value)
-}