Prevent local-activity panics from taking down the worker process (#1169)

Previously, a local activity that panicked would lead to a panic
exiting the goroutine that performed the call, terminating the whole
worker process immediately.

Obviously this is not good behavior.  So now that's fixed.

Seems like it has been this way since local activities were introduced,
in commit 79c072a , so it's pretty
surprising that it went unnoticed / unfixed for this long.
Particularly since it stands a decent chance of killing every worker
instance due to decision task retries.

Author: Groxx

internal/internal_task_pollers.go

@@ -602,25 +602,8 @@ func (lath *localActivityTaskHandler) executeLocalActivityTask(task *localActivi
 		}
 	}
 
-	// panic handler
+	// count all failures beyond this point, as they come from the activity itself
 	defer func() {
-		if p := recover(); p != nil {
-			topLine := fmt.Sprintf("local activity for %s [panic]:", activityType)
-			st := getStackTraceRaw(topLine, 7, 0)
-			lath.logger.Error("LocalActivity panic.",
-				zap.String(tagWorkflowID, task.params.WorkflowInfo.WorkflowExecution.ID),
-				zap.String(tagRunID, task.params.WorkflowInfo.WorkflowExecution.RunID),
-				zap.String(tagActivityType, activityType),
-				zap.String(tagPanicError, fmt.Sprintf("%v", p)),
-				zap.String(tagPanicStack, st))
-			metricsScope.Counter(metrics.LocalActivityPanicCounter).Inc(1)
-			panicErr := newPanicError(p, st)
-			result = &localActivityResult{
-				task:   task,
-				result: nil,
-				err:    panicErr,
-			}
-		}
 		if result.err != nil {
 			metricsScope.Counter(metrics.LocalActivityFailedCounter).Inc(1)
 		}
@@ -648,12 +631,28 @@ func (lath *localActivityTaskHandler) executeLocalActivityTask(task *localActivi
 	var err error
 	doneCh := make(chan struct{})
 	go func(ch chan struct{}) {
+		defer close(ch)
+
+		defer func() {
+			if p := recover(); p != nil {
+				topLine := fmt.Sprintf("local activity for %s [panic]:", activityType)
+				st := getStackTraceRaw(topLine, 7, 0)
+				lath.logger.Error("LocalActivity panic.",
+					zap.String(tagWorkflowID, task.params.WorkflowInfo.WorkflowExecution.ID),
+					zap.String(tagRunID, task.params.WorkflowInfo.WorkflowExecution.RunID),
+					zap.String(tagActivityType, activityType),
+					zap.String(tagPanicError, fmt.Sprintf("%v", p)),
+					zap.String(tagPanicStack, st))
+				metricsScope.Counter(metrics.LocalActivityPanicCounter).Inc(1)
+				err = newPanicError(p, st)
+			}
+		}()
+
 		laStartTime := time.Now()
 		ctx, span := createOpenTracingActivitySpan(ctx, lath.tracer, time.Now(), task.params.ActivityType, task.params.WorkflowInfo.WorkflowExecution.ID, task.params.WorkflowInfo.WorkflowExecution.RunID)
 		defer span.Finish()
 		laResult, err = ae.ExecuteWithActualArgs(ctx, task.params.InputArgs)
 		executionLatency := time.Now().Sub(laStartTime)
-		close(ch)
 		metricsScope.Timer(metrics.LocalActivityExecutionLatency).Record(executionLatency)
 		if executionLatency > timeoutDuration {
 			// If local activity takes longer than expected timeout, the context would already be DeadlineExceeded and

internal/internal_task_pollers_test.go

@@ -0,0 +1,36 @@
+package internal
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zaptest"
+)
+
+func TestLocalActivityPanic(t *testing.T) {
+	// regression: panics in local activities should not terminate the process
+	s := WorkflowTestSuite{logger: zaptest.NewLogger(t)}
+	env := s.NewTestWorkflowEnvironment()
+
+	wf := "panicky_local_activity"
+	env.RegisterWorkflowWithOptions(func(ctx Context) error {
+		ctx = WithLocalActivityOptions(ctx, LocalActivityOptions{
+			ScheduleToCloseTimeout: time.Second,
+		})
+		return ExecuteLocalActivity(ctx, func(ctx context.Context) error {
+			panic("should not kill process")
+		}).Get(ctx, nil)
+	}, RegisterWorkflowOptions{Name: wf})
+
+	env.ExecuteWorkflow(wf)
+	err := env.GetWorkflowError()
+	require.Error(t, err)
+	var perr *PanicError
+	require.True(t, errors.As(err, &perr), "error should be a panic error")
+	assert.Contains(t, perr.StackTrace(), "panic")
+	assert.Contains(t, perr.StackTrace(), t.Name(), "should mention the source location of the local activity that panicked")
+}