Allow replayer to use local activities with string names

internal/workflow.go

@@ -25,6 +25,7 @@
 package internal
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
@@ -629,11 +630,16 @@ func (wc *workflowEnvironmentInterceptor) ExecuteLocalActivity(ctx Context, type
 			return future
 		}
 		activity, ok := registry.GetActivity(activityType.Name)
-		if !ok {
+		if ok {
+			activityFn = activity.GetFunction()
+		} else if IsReplayNamespace(GetWorkflowInfo(ctx).Namespace) {
+			// When running the replayer (but not necessarily during all replays), we
+			// don't require the activities to be registered, so use a dummy function
+			activityFn = func(context.Context) error { panic("dummy replayer function") }
+		} else {
 			settable.Set(nil, fmt.Errorf("local activity %s is not registered by the worker", activityType.Name))
 			return future
 		}
-		activityFn = activity.GetFunction()
 	} else {
 		if err := validateFunctionArgs(localCtx.fn, args, false); err != nil {
 			settable.Set(nil, err)

test/integration_test.go

@@ -1940,6 +1940,33 @@ func (ts *IntegrationTestSuite) TestReturnCancelError() {
 	ts.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_CANCELED, resp.GetWorkflowExecutionInfo().GetStatus())
 }
 
+func (ts *IntegrationTestSuite) TestLocalActivityStringNameReplay() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Run the workflow
+	run, err := ts.client.ExecuteWorkflow(ctx,
+		ts.startWorkflowOptions("test-local-activity-string-name-replay"), ts.workflows.LocalActivityByStringName)
+	ts.NotNil(run)
+	ts.NoError(err)
+	ts.NoError(run.Get(ctx, nil))
+
+	// Obtain history
+	var history historypb.History
+	iter := ts.client.GetWorkflowHistory(ctx, run.GetID(), run.GetRunID(), false,
+		enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
+	for iter.HasNext() {
+		event, err := iter.Next()
+		ts.NoError(err)
+		history.Events = append(history.Events, event)
+	}
+
+	// Run in replayer
+	replayer := worker.NewWorkflowReplayer()
+	replayer.RegisterWorkflow(ts.workflows.LocalActivityByStringName)
+	ts.NoError(replayer.ReplayWorkflowHistory(nil, &history))
+}
+
 func (ts *IntegrationTestSuite) TestMaxConcurrentSessionExecutionSize() {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()

test/workflow_test.go

@@ -1724,6 +1724,11 @@ func (w *Workflows) ReturnCancelError(
 	return temporal.NewCanceledError("some details")
 }
 
+func (w *Workflows) LocalActivityByStringName(ctx workflow.Context) error {
+	ctx = workflow.WithLocalActivityOptions(ctx, w.defaultLocalActivityOptions())
+	return workflow.ExecuteLocalActivity(ctx, "Prefix_ToUpper", "somestring").Get(ctx, nil)
+}
+
 func (w *Workflows) register(worker worker.Worker) {
 	worker.RegisterWorkflow(w.ActivityCancelRepro)
 	worker.RegisterWorkflow(w.ActivityCompletionUsingID)
@@ -1791,6 +1796,7 @@ func (w *Workflows) register(worker worker.Worker) {
 	worker.RegisterWorkflow(w.TooFewParams)
 	worker.RegisterWorkflow(w.ExecuteRemoteActivityToUpper)
 	worker.RegisterWorkflow(w.ReturnCancelError)
+	worker.RegisterWorkflow(w.LocalActivityByStringName)
 
 	worker.RegisterWorkflow(w.child)
 	worker.RegisterWorkflow(w.childForMemoAndSearchAttr)