Service boilerplate

bergundy · bergundy · commit 1882ace752b4 · 2025-10-03T16:15:43.000-07:00
diff --git a/chasm/lib/activity/frontend.go b/chasm/lib/activity/frontend.go
@@ -0,0 +1,38 @@
+package activity
+
+import (
+	"context"
+
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
+)
+
+type FrontendHandler interface {
+	StartActivityExecution(ctx context.Context, req *workflowservice.StartActivityExecutionRequest) (*workflowservice.StartActivityExecutionResponse, error)
+	CountActivityExecutions(context.Context, *workflowservice.CountActivityExecutionsRequest) (*workflowservice.CountActivityExecutionsResponse, error)
+	DeleteActivityExecution(context.Context, *workflowservice.DeleteActivityExecutionRequest) (*workflowservice.DeleteActivityExecutionResponse, error)
+	DescribeActivityExecution(context.Context, *workflowservice.DescribeActivityExecutionRequest) (*workflowservice.DescribeActivityExecutionResponse, error)
+	GetActivityResult(context.Context, *workflowservice.GetActivityResultRequest) (*workflowservice.GetActivityResultResponse, error)
+	ListActivityExecutions(context.Context, *workflowservice.ListActivityExecutionsRequest) (*workflowservice.ListActivityExecutionsResponse, error)
+	RequestCancelActivityExecution(context.Context, *workflowservice.RequestCancelActivityExecutionRequest) (*workflowservice.RequestCancelActivityExecutionResponse, error)
+	TerminateActivityExecution(context.Context, *workflowservice.TerminateActivityExecutionRequest) (*workflowservice.TerminateActivityExecutionResponse, error)
+}
+
+type frontendHandler struct {
+	FrontendHandler
+	client activitypb.ActivityServiceClient
+}
+
+func NewFrontendHandler(client activitypb.ActivityServiceClient) FrontendHandler {
+	return &frontendHandler{
+		client: client,
+	}
+}
+
+func (h *frontendHandler) StartActivityExecution(ctx context.Context, req *workflowservice.StartActivityExecutionRequest) (*workflowservice.StartActivityExecutionResponse, error) {
+	resp, err := h.client.StartActivityExecution(ctx, &activitypb.StartActivityExecutionRequest{
+		// NamespaceId
+		FrontendRequest: req,
+	})
+	return resp.GetFrontendResponse(), err
+}
diff --git a/chasm/lib/activity/fx.go b/chasm/lib/activity/fx.go
@@ -0,0 +1,24 @@
+package activity
+
+import (
+	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
+	"go.uber.org/fx"
+)
+
+var HistoryModule = fx.Module(
+	"activity-history",
+	fx.Provide(
+		newHandler,
+		newLibrary,
+	),
+	fx.Invoke(func(l *library, registry *chasm.Registry) {
+		registry.Register(l)
+	}),
+)
+
+var FrontendModule = fx.Module(
+	"activity-frontend",
+	fx.Provide(activitypb.NewActivityServiceLayeredClient),
+	fx.Provide(NewFrontendHandler),
+)
diff --git a/chasm/lib/activity/gen/activitypb/v1/service_client.pb.go b/chasm/lib/activity/gen/activitypb/v1/service_client.pb.go
diff --git a/chasm/lib/activity/handler.go b/chasm/lib/activity/handler.go
@@ -0,0 +1,24 @@
+package activity
+
+import (
+	"context"
+
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
+)
+
+type handler struct {
+	activitypb.UnimplementedActivityServiceServer
+}
+
+func newHandler() *handler {
+	return &handler{}
+}
+
+func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.StartActivityExecutionRequest) (*activitypb.StartActivityExecutionResponse, error) {
+	return &activitypb.StartActivityExecutionResponse{
+		FrontendResponse: &workflowservice.StartActivityExecutionResponse{
+			// TODO
+		},
+	}, nil
+}
diff --git a/chasm/lib/activity/library.go b/chasm/lib/activity/library.go
@@ -0,0 +1,24 @@
+package activity
+
+import (
+	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
+	"google.golang.org/grpc"
+)
+
+type library struct {
+	chasm.UnimplementedLibrary
+	handler *handler
+}
+
+func newLibrary(*handler) *library {
+	return &library{}
+}
+
+func (l *library) Name() string {
+	return "activity"
+}
+
+func (l *library) RegisterServices(server *grpc.Server) {
+	server.RegisterService(&activitypb.ActivityService_ServiceDesc, l.handler)
+}
diff --git a/chasm/lib/tests/gen/testspb/v1/service_client.pb.go b/chasm/lib/tests/gen/testspb/v1/service_client.pb.go
diff --git a/cmd/tools/protoc-gen-go-chasm/main.go b/cmd/tools/protoc-gen-go-chasm/main.go
@@ -192,7 +192,7 @@ func (p *Plugin) genClient(w *writer, svc *protogen.Service) error {
 
 	ctorName := fmt.Sprintf("New%s", structName)
 	w.println("// %s initializes a new %s.", ctorName, structName)
-	w.println("func New%s(", ctorName)
+	w.println("func %s(", ctorName)
 	w.indent()
 	w.println("dc *dynamicconfig.Collection,")
 	w.println("rpcFactory     common.RPCFactory,")
@@ -201,7 +201,7 @@ func (p *Plugin) genClient(w *writer, svc *protogen.Service) error {
 	w.println("logger         log.Logger,")
 	w.println("metricsHandler metrics.Handler,")
 	w.unindent()
-	w.println(") (*%s, error) {", structName)
+	w.println(") (%sClient, error) {", svc.GoName)
 	w.indent() // start ctor body
 	w.println("resolver, err := monitor.GetResolver(primitives.HistoryService)")
 	w.println("if err != nil {")
diff --git a/common/rpc/interceptor/redirection.go b/common/rpc/interceptor/redirection.go
@@ -135,6 +135,15 @@ var (
 		"UpdateTaskQueueConfig": func() any { return &workflowservice.UpdateTaskQueueConfigResponse{} },
 		"FetchWorkerConfig":     func() any { return &workflowservice.FetchWorkerConfigResponse{} },
 		"UpdateWorkerConfig":    func() any { return &workflowservice.UpdateWorkerConfigResponse{} },
+
+		"StartActivityExecution":         func() any { return &workflowservice.StartActivityExecutionResponse{} },
+		"CountActivityExecutions":        func() any { return &workflowservice.CountActivityExecutionsResponse{} },
+		"ListActivityExecutions":         func() any { return &workflowservice.ListActivityExecutionsResponse{} },
+		"DescribeActivityExecution":      func() any { return &workflowservice.DescribeActivityExecutionResponse{} },
+		"GetActivityResult":              func() any { return &workflowservice.GetActivityResultResponse{} },
+		"RequestCancelActivityExecution": func() any { return &workflowservice.RequestCancelActivityExecutionResponse{} },
+		"TerminateActivityExecution":     func() any { return &workflowservice.TerminateActivityExecutionResponse{} },
+		"DeleteActivityExecution":        func() any { return &workflowservice.DeleteActivityExecutionResponse{} },
 	}
 )
 
diff --git a/common/rpc/interceptor/redirection_test.go b/common/rpc/interceptor/redirection_test.go
@@ -190,6 +190,15 @@ func (s *redirectionInterceptorSuite) TestGlobalAPI() {
 		"UpdateTaskQueueConfig":                 {},
 		"FetchWorkerConfig":                     {},
 		"UpdateWorkerConfig":                    {},
+
+		"StartActivityExecution":         {},
+		"CountActivityExecutions":        {},
+		"ListActivityExecutions":         {},
+		"DescribeActivityExecution":      {},
+		"GetActivityResult":              {},
+		"RequestCancelActivityExecution": {},
+		"TerminateActivityExecution":     {},
+		"DeleteActivityExecution":        {},
 	}, apis)
 }
 
diff --git a/service/frontend/configs/quotas.go b/service/frontend/configs/quotas.go
@@ -42,6 +42,8 @@ var (
 		"/temporal.api.workflowservice.v1.WorkflowService/UpdateWorkflowExecution":     1,
 		"/temporal.api.workflowservice.v1.WorkflowService/GetWorkflowExecutionHistory": 1,
 		"/temporal.api.workflowservice.v1.WorkflowService/PollNexusTaskQueue":          1,
+		// TODO: Map to PollActivityResult if request is long-polling
+		"/temporal.api.workflowservice.v1.WorkflowService/GetActivityResult": 1,
 
 		// potentially long-running, depending on the operations
 		"/temporal.api.workflowservice.v1.WorkflowService/ExecuteMultiOperation": 1,
@@ -71,6 +73,7 @@ var (
 		"/temporal.api.workflowservice.v1.WorkflowService/ExecuteMultiOperation":            1,
 		"/temporal.api.workflowservice.v1.WorkflowService/CreateSchedule":                   1,
 		"/temporal.api.workflowservice.v1.WorkflowService/StartBatchOperation":              1,
+		"/temporal.api.workflowservice.v1.WorkflowService/StartActivityExecution":           1,
 		DispatchNexusTaskByNamespaceAndTaskQueueAPIName:                                     1,
 		DispatchNexusTaskByEndpointAPIName:                                                  1,
 
@@ -114,6 +117,10 @@ var (
 		"/temporal.api.workflowservice.v1.WorkflowService/ListWorkflowRules":                     2,
 		"/temporal.api.workflowservice.v1.WorkflowService/TriggerWorkflowRule":                   2,
 		"/temporal.api.workflowservice.v1.WorkflowService/UpdateTaskQueueConfig":                 2,
+		"/temporal.api.workflowservice.v1.WorkflowService/RequestCancelActivityExecution":        2,
+		"/temporal.api.workflowservice.v1.WorkflowService/TerminateActivityExecution":            2,
+		"/temporal.api.workflowservice.v1.WorkflowService/DeleteActivityExecution":               2,
+		"/temporal.api.workflowservice.v1.WorkflowService/GetActivityResult":                     2,
 
 		// P3: Status Querying APIs
 		"/temporal.api.workflowservice.v1.WorkflowService/DescribeWorkflowExecution":       3,
@@ -130,6 +137,7 @@ var (
 		"/temporal.api.workflowservice.v1.WorkflowService/DescribeWorkerDeploymentVersion": 3,
 		"/temporal.api.workflowservice.v1.WorkflowService/DescribeWorkerDeployment":        3,
 		"/temporal.api.workflowservice.v1.WorkflowService/ListWorkerDeployments":           3,
+		"/temporal.api.workflowservice.v1.WorkflowService/DescribeActivityExecution":       3,
 
 		// P3: Progress APIs for reporting cancellations and failures.
 		// They are relatively low priority as the tasks need to be retried anyway.
@@ -171,6 +179,8 @@ var (
 		"/temporal.api.workflowservice.v1.WorkflowService/ListArchivedWorkflowExecutions": 1,
 		"/temporal.api.workflowservice.v1.WorkflowService/ListWorkers":                    1,
 		"/temporal.api.workflowservice.v1.WorkflowService/DescribeWorker":                 1,
+		"/temporal.api.workflowservice.v1.WorkflowService/CountActivityExecutions":        1,
+		"/temporal.api.workflowservice.v1.WorkflowService/ListActivityExecutions":         1,
 
 		// APIs that rely on visibility
 		"/temporal.api.workflowservice.v1.WorkflowService/GetWorkerTaskReachability":         1,
diff --git a/service/frontend/configs/quotas_test.go b/service/frontend/configs/quotas_test.go
@@ -101,6 +101,9 @@ func (s *quotasSuite) TestVisibilityAPIs() {
 		"/temporal.api.workflowservice.v1.WorkflowService/DescribeTaskQueueWithReachability": {},
 		"/temporal.api.workflowservice.v1.WorkflowService/ListDeployments":                   {},
 		"/temporal.api.workflowservice.v1.WorkflowService/GetDeploymentReachability":         {},
+
+		"/temporal.api.workflowservice.v1.WorkflowService/CountActivityExecutions": {},
+		"/temporal.api.workflowservice.v1.WorkflowService/ListActivityExecutions":  {},
 	}
 
 	var service workflowservice.WorkflowServiceServer
diff --git a/service/frontend/fx.go b/service/frontend/fx.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/gorilla/mux"
 	"go.temporal.io/server/api/adminservice/v1"
+	"go.temporal.io/server/chasm/lib/activity"
 	"go.temporal.io/server/client"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/archiver"
@@ -109,6 +110,7 @@ var Module = fx.Options(
 	fx.Invoke(ServiceLifetimeHooks),
 	fx.Invoke(EndpointRegistryLifetimeHooks),
 	nexusfrontend.Module,
+	activity.FrontendModule,
 )
 
 func NewServiceProvider(
@@ -731,6 +733,7 @@ func HandlerProvider(
 	membershipMonitor membership.Monitor,
 	healthInterceptor *interceptor.HealthInterceptor,
 	scheduleSpecBuilder *scheduler.SpecBuilder,
+	activityHandler activity.FrontendHandler,
 ) Handler {
 	wfHandler := NewWorkflowHandler(
 		serviceConfig,
@@ -758,6 +761,7 @@ func HandlerProvider(
 		healthInterceptor,
 		scheduleSpecBuilder,
 		httpEnabled(cfg, serviceName),
+		activityHandler,
 	)
 	return wfHandler
 }
diff --git a/service/frontend/workflow_handler.go b/service/frontend/workflow_handler.go
@@ -32,6 +32,7 @@ import (
 	"go.temporal.io/server/api/matchingservice/v1"
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
+	"go.temporal.io/server/chasm/lib/activity"
 	"go.temporal.io/server/client/frontend"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/archiver"
@@ -110,7 +111,9 @@ const (
 type (
 	// WorkflowHandler - gRPC handler interface for workflowservice
 	WorkflowHandler struct {
-		workflowservice.UnimplementedWorkflowServiceServer
+		workflowservice.UnsafeWorkflowServiceServer
+		activity.FrontendHandler
+
 		status int32
 
 		tokenSerializer                 *tasktoken.Serializer
@@ -172,8 +175,10 @@ func NewWorkflowHandler(
 	healthInterceptor *interceptor.HealthInterceptor,
 	scheduleSpecBuilder *scheduler.SpecBuilder,
 	httpEnabled bool,
+	activityHandler activity.FrontendHandler,
 ) *WorkflowHandler {
 	handler := &WorkflowHandler{
+		FrontendHandler: activityHandler,
 		status:          common.DaemonStatusInitialized,
 		config:          config,
 		tokenSerializer: tasktoken.NewSerializer(),
@@ -6104,3 +6109,7 @@ func (wh *WorkflowHandler) DescribeWorker(ctx context.Context, request *workflow
 		WorkerInfo: resp.GetWorkerInfo(),
 	}, nil
 }
+
+func (wh *WorkflowHandler) TriggerWorkflowRule(context.Context, *workflowservice.TriggerWorkflowRuleRequest) (*workflowservice.TriggerWorkflowRuleResponse, error) {
+	return nil, serviceerror.NewUnimplemented("method TriggerWorkflowRule not supported")
+}
diff --git a/service/frontend/workflow_handler_test.go b/service/frontend/workflow_handler_test.go
@@ -189,6 +189,7 @@ func (s *WorkflowHandlerSuite) getWorkflowHandler(config *Config) *WorkflowHandl
 		healthInterceptor,
 		scheduler.NewSpecBuilder(),
 		true,
+		nil, // Not testing activity handler here
 	)
 }
 
diff --git a/service/history/fx.go b/service/history/fx.go
@@ -3,6 +3,7 @@ package history
 import (
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/chasm/lib/activity"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/config"
@@ -78,6 +79,7 @@ var Module = fx.Options(
 	callbacks.Module,
 	nexusoperations.Module,
 	fx.Invoke(nexusworkflow.RegisterCommandHandlers),
+	activity.HistoryModule,
 )
 
 func ServerProvider(grpcServerOptions []grpc.ServerOption) *grpc.Server {
diff --git a/tests/standalone_activity_test.go b/tests/standalone_activity_test.go
@@ -0,0 +1,42 @@
+package tests
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+	activitypb "go.temporal.io/api/activity/v1"
+	taskqueuepb "go.temporal.io/api/taskqueue/v1"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/tests/testcore"
+	"google.golang.org/protobuf/types/known/durationpb"
+)
+
+type standaloneActivityTestSuite struct {
+	testcore.FunctionalTestBase
+}
+
+func TestStandaloneActivityTestSuite(t *testing.T) {
+	t.Parallel()
+	suite.Run(t, new(standaloneActivityTestSuite))
+}
+
+func (s *standaloneActivityTestSuite) TestStartActivityExecution() {
+	t := s.T()
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+
+	_, err := s.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{
+		Namespace:  s.Namespace().String(),
+		ActivityId: testcore.RandomizeStr(t.Name()),
+		Options: &activitypb.ActivityOptions{
+			TaskQueue: &taskqueuepb.TaskQueue{
+				Name: s.TaskQueue(),
+			},
+			StartToCloseTimeout: durationpb.New(1 * time.Minute),
+		},
+	})
+	require.NoError(t, err)
+}
