Expose replicateWorkflowState API in history (#3783)

* Expose replicateWorkflowState API in history
temporalio · Jan 13, 2023 · 0e4d618 · 0e4d618
1 parent d321415
commit 0e4d618
Show file tree

Hide file tree

Showing 11 changed files with 822 additions and 432 deletions.
diff --git a/api/historyservice/v1/request_response.pb.go b/api/historyservice/v1/request_response.pb.go
diff --git a/api/historyservice/v1/service.pb.go b/api/historyservice/v1/service.pb.go
diff --git a/api/historyservicemock/v1/service.pb.mock.go b/api/historyservicemock/v1/service.pb.mock.go
diff --git a/client/history/client_gen.go b/client/history/client_gen.go
diff --git a/client/history/metric_client_gen.go b/client/history/metric_client_gen.go
diff --git a/client/history/retryable_client_gen.go b/client/history/retryable_client_gen.go
diff --git a/common/metrics/metric_defs.go b/common/metrics/metric_defs.go
@@ -495,6 +495,8 @@ const (
 	HistoryClientVerifyChildExecutionCompletionRecordedScope = "HistoryClientVerifyChildExecutionCompletionRecorded"
 	// HistoryClientReplicateEventsV2Scope tracks RPC calls to history service
 	HistoryClientReplicateEventsV2Scope = "HistoryClientReplicateEventsV2"
+	// HistoryClientReplicateWorkflowStateScope tracks RPC calls to history service
+	HistoryClientReplicateWorkflowStateScope = "HistoryClientReplicateWorkflowState"
 	// HistoryClientSyncShardStatusScope tracks RPC calls to history service
 	HistoryClientSyncShardStatusScope = "HistoryClientSyncShardStatus"
 	// HistoryClientSyncActivityScope tracks RPC calls to history service

diff --git a/proto/internal/temporal/server/api/historyservice/v1/request_response.proto b/proto/internal/temporal/server/api/historyservice/v1/request_response.proto
@@ -382,12 +382,16 @@ message ReplicateEventsV2Request {
     temporal.api.common.v1.DataBlob new_run_events = 5;
 }
 
+message ReplicateEventsV2Response {
+}
+
 message ReplicateWorkflowStateRequest {
     temporal.server.api.persistence.v1.WorkflowMutableState workflow_state = 1;
     string remote_cluster = 2;
+    string namespace_id= 3;
 }
 
-message ReplicateEventsV2Response {
+message ReplicateWorkflowStateResponse {
 }
 
 message SyncShardStatusRequest {

diff --git a/proto/internal/temporal/server/api/historyservice/v1/service.proto b/proto/internal/temporal/server/api/historyservice/v1/service.proto
@@ -192,9 +192,14 @@ service HistoryService {
     rpc DescribeWorkflowExecution (DescribeWorkflowExecutionRequest) returns (DescribeWorkflowExecutionResponse) {
     }
 
+    // ReplicateEventsV2 replicates workflow history events
     rpc ReplicateEventsV2 (ReplicateEventsV2Request) returns (ReplicateEventsV2Response) {
     }
 
+    // ReplicateWorkflowState replicates workflow state
+    rpc ReplicateWorkflowState(ReplicateWorkflowStateRequest) returns (ReplicateWorkflowStateResponse) {
+    }
+
     // SyncShardStatus sync the status between shards.
     rpc SyncShardStatus (SyncShardStatusRequest) returns (SyncShardStatusResponse) {
     }
@@ -275,7 +280,7 @@ service HistoryService {
     }
 
     // (-- api-linter: core::0134=disabled
-    //     aip.dev/not-precedent: This service does not follow the update method AIP --)
+    //     aip.dev/not-precedent: This service does not follow the update method API --)
     rpc UpdateWorkflow(UpdateWorkflowRequest) returns (UpdateWorkflowResponse) {
     }
 }
diff --git a/service/history/configs/quotas.go b/service/history/configs/quotas.go
@@ -55,6 +55,7 @@ var (
 		"RemoveSignalMutableState":               0,
 		"RemoveTask":                             0,
 		"ReplicateEventsV2":                      0,
+		"ReplicateWorkflowState":                 0,
 		"RequestCancelWorkflowExecution":         0,
 		"ResetStickyTaskQueue":                   0,
 		"ResetWorkflowExecution":                 0,

diff --git a/service/history/handler.go b/service/history/handler.go
@@ -1301,6 +1301,38 @@ func (h *Handler) ReplicateEventsV2(ctx context.Context, request *historyservice
 	return &historyservice.ReplicateEventsV2Response{}, nil
 }
 
+// ReplicateWorkflowState is called by processor to replicate workflow state for passive namespaces
+func (h *Handler) ReplicateWorkflowState(
+	ctx context.Context,
+	request *historyservice.ReplicateWorkflowStateRequest,
+) (_ *historyservice.ReplicateWorkflowStateResponse, retError error) {
+	defer log.CapturePanic(h.logger, &retError)
+	h.startWG.Wait()
+
+	if h.isStopped() {
+		return nil, errShuttingDown
+	}
+
+	shardContext, err := h.controller.GetShardByNamespaceWorkflow(
+		namespace.ID(request.GetWorkflowState().GetExecutionInfo().GetNamespaceId()),
+		request.GetWorkflowState().GetExecutionInfo().GetWorkflowId(),
+	)
+	if err != nil {
+		return nil, h.convertError(err)
+	}
+
+	engine, err := shardContext.GetEngine(ctx)
+	if err != nil {
+		return nil, h.convertError(err)
+	}
+
+	err = engine.ReplicateWorkflowState(ctx, request)
+	if err != nil {
+		return nil, err
+	}
+	return &historyservice.ReplicateWorkflowStateResponse{}, nil
+}
+
 // SyncShardStatus is called by processor to sync history shard information from another cluster
 func (h *Handler) SyncShardStatus(ctx context.Context, request *historyservice.SyncShardStatusRequest) (_ *historyservice.SyncShardStatusResponse, retError error) {
 	defer log.CapturePanic(h.logger, &retError)