improvement: bypass middle-man-struct for batch operations

service/frontend/workflow_handler.go

@@ -4505,7 +4505,7 @@
 	var resetParams batcher.ResetParams
 	var updateOptionsParams batcher.UpdateOptionsParams
 	var unpauseActivitiesParams batcher.UnpauseActivitiesParams
-	var resetActivitiesParams batcher.ResetActivitiesParams
+	var resetActivitiesParams *workflowservice.StartBatchOperationRequest_ResetActivitiesOperation
 	var updateActivitiesOptionsParams batcher.UpdateActivitiesOptionsParams
 	switch op := request.Operation.(type) {
 	case *workflowservice.StartBatchOperationRequest_TerminationOperation:
@@ -4601,7 +4601,7 @@
 		unpauseActivitiesParams.Identity = op.UnpauseActivitiesOperation.GetIdentity()
 	case *workflowservice.StartBatchOperationRequest_ResetActivitiesOperation:
 		operationType = batcher.BatchTypeResetActivities
-		if op.ResetActivitiesOperation == nil {
+		if op.ResetActivitiesOperation == nil {			
 			return nil, serviceerror.NewInvalidArgument("reset activities operation is not set")
 		}
 		if op.ResetActivitiesOperation.GetActivity() == nil {
@@ -4613,20 +4613,20 @@
 			if len(a.Type) == 0 {
 				return nil, serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
 			}
-			resetActivitiesParams.ActivityType = a.Type
+			resetActivitiesParams.ResetActivitiesOperation.Activity = a
 		case *batchpb.BatchOperationResetActivities_MatchAll:
 			if !a.MatchAll {
 				return nil, serviceerror.NewInvalidArgument("Either activity type must be set, or match all should be set to true")
 			}
-			resetActivitiesParams.MatchAll = true
+			resetActivitiesParams.ResetActivitiesOperation.Activity = a
 		}
 
-		resetActivitiesParams.ResetAttempts = op.ResetActivitiesOperation.ResetAttempts
-		resetActivitiesParams.ResetHeartbeat = op.ResetActivitiesOperation.ResetHeartbeat
-		resetActivitiesParams.Jitter = op.ResetActivitiesOperation.Jitter.AsDuration()
-		resetActivitiesParams.KeepPaused = op.ResetActivitiesOperation.KeepPaused
-		resetActivitiesParams.RestoreOriginalOptions = op.ResetActivitiesOperation.RestoreOriginalOptions
-		resetActivitiesParams.Identity = op.ResetActivitiesOperation.GetIdentity()
+		resetActivitiesParams.ResetActivitiesOperation.ResetAttempts = op.ResetActivitiesOperation.ResetAttempts
+		resetActivitiesParams.ResetActivitiesOperation.ResetHeartbeat = op.ResetActivitiesOperation.ResetHeartbeat
+		resetActivitiesParams.ResetActivitiesOperation.Jitter = op.ResetActivitiesOperation.Jitter
+		resetActivitiesParams.ResetActivitiesOperation.KeepPaused = op.ResetActivitiesOperation.KeepPaused
+		resetActivitiesParams.ResetActivitiesOperation.RestoreOriginalOptions = op.ResetActivitiesOperation.RestoreOriginalOptions
+		resetActivitiesParams.ResetActivitiesOperation.Identity = op.ResetActivitiesOperation.GetIdentity()
 	case *workflowservice.StartBatchOperationRequest_UpdateActivityOptionsOperation:
 		operationType = batcher.BatchTypeUpdateActivitiesOptions
 		if op.UpdateActivityOptionsOperation == nil {

service/worker/batcher/activities.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/pborman/uuid"
 	activitypb "go.temporal.io/api/activity/v1"
+	batchpb "go.temporal.io/api/batch/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
@@ -382,18 +383,21 @@ func startTaskProcessor(
 								WorkflowId: workflowID,
 								RunId:      runID,
 							},
-							Identity:               batchParams.ResetActivitiesParams.Identity,
-							Activity:               &workflowservice.ResetActivityRequest_Type{Type: batchParams.ResetActivitiesParams.ActivityType},
-							ResetHeartbeat:         batchParams.ResetActivitiesParams.ResetHeartbeat,
-							Jitter:                 durationpb.New(batchParams.ResetActivitiesParams.Jitter),
-							KeepPaused:             batchParams.ResetActivitiesParams.KeepPaused,
-							RestoreOriginalOptions: batchParams.ResetActivitiesParams.RestoreOriginalOptions,
+							Identity:               batchParams.ResetActivitiesParams.ResetActivitiesOperation.Identity,
+							Activity:               &workflowservice.ResetActivityRequest_Type{Type: batchParams.ResetActivitiesParams.ResetActivitiesOperation.GetType()},
+							ResetHeartbeat:         batchParams.ResetActivitiesParams.ResetActivitiesOperation.ResetHeartbeat,
+							Jitter:                 batchParams.ResetActivitiesParams.ResetActivitiesOperation.Jitter,
+							KeepPaused:             batchParams.ResetActivitiesParams.ResetActivitiesOperation.KeepPaused,
+							RestoreOriginalOptions: batchParams.ResetActivitiesParams.ResetActivitiesOperation.RestoreOriginalOptions,
 						}
 
-						if batchParams.ResetActivitiesParams.MatchAll {
+						switch a := batchParams.ResetActivitiesParams.ResetActivitiesOperation.GetActivity().(type) {
+						case *batchpb.BatchOperationResetActivities_Type:
+							resetRequest.Activity = &workflowservice.ResetActivityRequest_Type{Type: a.Type}
+						case *batchpb.BatchOperationResetActivities_MatchAll:
 							resetRequest.Activity = &workflowservice.ResetActivityRequest_MatchAll{MatchAll: true}
-						} else {
-							resetRequest.Activity = &workflowservice.ResetActivityRequest_Type{Type: batchParams.ResetActivitiesParams.ActivityType}
+						default:
+							return fmt.Errorf("invalid activity type: %T", a)
 						}
 
 						_, err = frontendClient.ResetActivity(ctx, resetRequest)

service/worker/batcher/workflow.go

@@ -5,9 +5,11 @@
 	"fmt"
 	"time"
 
+	batchpb "go.temporal.io/api/batch/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	workflowpb "go.temporal.io/api/workflow/v1"
+	workflowservicepb "go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/workflow"
 	"go.temporal.io/server/common/searchattribute"
@@ -139,17 +141,6 @@
 		NonRetryableErrorTypes []string
 	}
 
-	ResetActivitiesParams struct {
-		Identity               string
-		ActivityType           string
-		MatchAll               bool
-		ResetAttempts          bool
-		ResetHeartbeat         bool
-		KeepPaused             bool
-		Jitter                 time.Duration
-		RestoreOriginalOptions bool
-	}
-
 	FieldMask struct {
 		Paths []string
 	}
@@ -185,7 +176,7 @@
 		// UpdateActivitiesOptionsParams is params only for BatchTypeUpdateActivitiesOptions
 		UpdateActivitiesOptionsParams UpdateActivitiesOptionsParams
 		// ResetActivitiesParams is params only for BatchTypeResetActivities
-		ResetActivitiesParams ResetActivitiesParams
+		ResetActivitiesParams *workflowservicepb.StartBatchOperationRequest_ResetActivitiesOperation
 
 		// RPS sets the requests-per-second limit for the batch.
 		// The default (and max) is defined by `worker.BatcherRPS` in the dynamic config.
@@ -313,8 +304,19 @@
 		}
 		return nil
 	case BatchTypeResetActivities:
-		if params.ResetActivitiesParams.ActivityType == "" && !params.ResetActivitiesParams.MatchAll {
-			return errors.New("must provide ActivityType or MatchAll")
+		if params.ResetActivitiesParams.ResetActivitiesOperation.GetActivity() == nil {
+			return fmt.Errorf("must provide ActivityType or MatchAll")
+		}
+
+		switch a := params.ResetActivitiesParams.ResetActivitiesOperation.GetActivity().(type) {
+		case *batchpb.BatchOperationResetActivities_Type:
+			if len(a.Type) == 0 {
+				return fmt.Errorf("must provide ActivityType")
+			}
+		case *batchpb.BatchOperationResetActivities_MatchAll:
+			if !a.MatchAll {
+				return fmt.Errorf("must provide MatchAll")
+			}
 		}
 		return nil
 	case BatchTypeUpdateActivitiesOptions: