Temporary fix for WorkflowMutableState serialization

common/persistence/executionStore.go

@@ -110,17 +110,17 @@ func (m *executionManagerImpl) DeserializeExecutionInfo(
 	info *WorkflowExecutionInfo,
 ) (*WorkflowExecutionInfo, *persistenceblobs.ExecutionStats, error) {
 	newInfo := &WorkflowExecutionInfo{
-		ExecutionState:         info.ExecutionState,
-		CompletionEvent:        info.CompletionEvent,
-		NamespaceId:            info.NamespaceId,
-		WorkflowId:             info.WorkflowId,
-		FirstExecutionRunId:    info.FirstExecutionRunId,
-		ParentNamespaceId:      info.ParentNamespaceId,
-		ParentWorkflowId:       info.ParentWorkflowId,
-		ParentRunId:            info.ParentRunId,
-		InitiatedId:            info.InitiatedId,
-		CompletionEventBatchId: info.CompletionEventBatchId,
-		TaskQueue:              info.TaskQueue,
+		ExecutionState:                         info.ExecutionState,
+		CompletionEvent:                        info.CompletionEvent,
+		NamespaceId:                            info.NamespaceId,
+		WorkflowId:                             info.WorkflowId,
+		FirstExecutionRunId:                    info.FirstExecutionRunId,
+		ParentNamespaceId:                      info.ParentNamespaceId,
+		ParentWorkflowId:                       info.ParentWorkflowId,
+		ParentRunId:                            info.ParentRunId,
+		InitiatedId:                            info.InitiatedId,
+		CompletionEventBatchId:                 info.CompletionEventBatchId,
+		TaskQueue:                              info.TaskQueue,
 		WorkflowTypeName:                       info.WorkflowTypeName,
 		WorkflowRunTimeout:                     info.WorkflowRunTimeout,
 		WorkflowExecutionTimeout:               info.WorkflowExecutionTimeout,
@@ -235,27 +235,27 @@ func (m *executionManagerImpl) SerializeExecutionInfo(
 		WorkflowId:                             info.WorkflowId,
 		FirstExecutionRunId:                    info.FirstExecutionRunId,
 		ParentNamespaceId:                      info.ParentNamespaceId,
-		ParentWorkflowId:           info.ParentWorkflowId,
-		ParentRunId:                info.ParentRunId,
-		InitiatedId:                info.InitiatedId,
-		CompletionEventBatchId:     info.CompletionEventBatchId,
-		CompletionEvent:            info.CompletionEvent,
-		TaskQueue:                  info.TaskQueue,
-		WorkflowTypeName:           info.WorkflowTypeName,
-		WorkflowRunTimeout:         info.WorkflowRunTimeout,
-		WorkflowExecutionTimeout:   info.WorkflowExecutionTimeout,
-		DefaultWorkflowTaskTimeout: info.DefaultWorkflowTaskTimeout,
-		ExecutionState:             info.ExecutionState,
-		LastFirstEventId:           info.LastFirstEventId,
-		LastEventTaskId:            info.LastEventTaskId,
-		NextEventId:                info.NextEventId,
-		LastProcessedEvent:         info.LastProcessedEvent,
-		StartTime:                  info.StartTime,
-		LastUpdatedTime:            info.LastUpdatedTime,
-		SignalCount:                info.SignalCount,
-		WorkflowTaskVersion:        info.WorkflowTaskVersion,
-		WorkflowTaskScheduleId:     info.WorkflowTaskScheduleId,
-		WorkflowTaskStartedId:      info.WorkflowTaskStartedId,
+		ParentWorkflowId:                       info.ParentWorkflowId,
+		ParentRunId:                            info.ParentRunId,
+		InitiatedId:                            info.InitiatedId,
+		CompletionEventBatchId:                 info.CompletionEventBatchId,
+		CompletionEvent:                        info.CompletionEvent,
+		TaskQueue:                              info.TaskQueue,
+		WorkflowTypeName:                       info.WorkflowTypeName,
+		WorkflowRunTimeout:                     info.WorkflowRunTimeout,
+		WorkflowExecutionTimeout:               info.WorkflowExecutionTimeout,
+		DefaultWorkflowTaskTimeout:             info.DefaultWorkflowTaskTimeout,
+		ExecutionState:                         info.ExecutionState,
+		LastFirstEventId:                       info.LastFirstEventId,
+		LastEventTaskId:                        info.LastEventTaskId,
+		NextEventId:                            info.NextEventId,
+		LastProcessedEvent:                     info.LastProcessedEvent,
+		StartTime:                              info.StartTime,
+		LastUpdatedTime:                        info.LastUpdatedTime,
+		SignalCount:                            info.SignalCount,
+		WorkflowTaskVersion:                    info.WorkflowTaskVersion,
+		WorkflowTaskScheduleId:                 info.WorkflowTaskScheduleId,
+		WorkflowTaskStartedId:                  info.WorkflowTaskStartedId,
 		WorkflowTaskRequestId:                  info.WorkflowTaskRequestId,
 		WorkflowTaskTimeout:                    info.WorkflowTaskTimeout,
 		WorkflowTaskAttempt:                    info.WorkflowTaskAttempt,

common/persistence/persistenceInterface.go

@@ -203,18 +203,17 @@ type (
 
 	// InternalWorkflowMutableState indicates workflow related state for Persistence Interface
 	InternalWorkflowMutableState struct {
-		ExecutionInfo    *WorkflowExecutionInfo
-		VersionHistories *history.VersionHistories
-		ActivityInfos    map[int64]*persistenceblobs.ActivityInfo
-
+		ActivityInfos       map[int64]*persistenceblobs.ActivityInfo
 		TimerInfos          map[string]*persistenceblobs.TimerInfo
 		ChildExecutionInfos map[int64]*persistenceblobs.ChildExecutionInfo
 		RequestCancelInfos  map[int64]*persistenceblobs.RequestCancelInfo
 		SignalInfos         map[int64]*persistenceblobs.SignalInfo
 		SignalRequestedIDs  map[string]struct{}
-		BufferedEvents      []*serialization.DataBlob
+		ExecutionInfo       *WorkflowExecutionInfo
 
-		Checksum checksum.Checksum
+		BufferedEvents   []*serialization.DataBlob
+		VersionHistories *history.VersionHistories
+		Checksum         checksum.Checksum
 	}
 
 	// InternalUpdateWorkflowExecutionRequest is used to update a workflow execution for Persistence Interface

proto/internal/temporal/server/api/adminservice/v1/request_response.proto

@@ -48,6 +48,8 @@ message DescribeWorkflowExecutionResponse {
     string history_addr = 2;
     string cache_mutable_state = 3;
     string database_mutable_state = 4;
+    string tree_id = 5;
+    string branch_id = 6;
 }
 
 //At least one of the parameters needs to be provided

proto/internal/temporal/server/api/historyservice/v1/request_response.proto

@@ -373,6 +373,8 @@ message DescribeMutableStateRequest {
 message DescribeMutableStateResponse {
     string cache_mutable_state = 1;
     string database_mutable_state = 2;
+    string tree_id = 3;
+    string branch_id = 4;
 }
 
 //At least one of the parameters needs to be provided

schema/cassandra/temporal/schema.cql

@@ -7,7 +7,7 @@ CREATE TYPE serialized_event_batch (
 CREATE TABLE executions (
   shard_id                       int,
   type                           int, -- enum RowType { Shard, Execution, TransferTask, TimerTask, ReplicationTask}
-  namespace_id                      uuid,
+  namespace_id                   uuid,
   workflow_id                    text,
   run_id                         uuid,
   current_run_id                 uuid,
@@ -26,7 +26,7 @@ CREATE TABLE executions (
   timer                          blob,
   timer_encoding                 text,
   next_event_id                  bigint,  -- This is needed to make conditional updates on session history
-  range_id                       bigint, -- Increasing sequence identifier for transfer queue, checkpointed into shard info
+  range_id                       bigint,  -- Increasing sequence identifier for transfer queue, checkpointed into shard info
   activity_map                   map<bigint, blob>,
   activity_map_encoding          text,
   timer_map                      map<text, blob>,

service/frontend/adminHandler.go

@@ -226,8 +226,10 @@ func (adh *AdminHandler) DescribeWorkflowExecution(ctx context.Context, request
 	return &adminservice.DescribeWorkflowExecutionResponse{
 		ShardId:              shardIDForOutput,
 		HistoryAddr:          historyAddr,
-		DatabaseMutableState: resp2.DatabaseMutableState,
-		CacheMutableState:    resp2.CacheMutableState,
+		DatabaseMutableState: resp2.GetDatabaseMutableState(),
+		CacheMutableState:    resp2.GetCacheMutableState(),
+		TreeId:               resp2.GetTreeId(),
+		BranchId:             resp2.GetBranchId(),
 	}, err
 }
 

service/history/historyEngine.go

@@ -29,7 +29,6 @@ package history
 import (
 	"bytes"
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"time"
@@ -48,6 +47,7 @@ import (
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/matchingservice/v1"
+	"go.temporal.io/server/api/persistenceblobs/v1"
 	replicationspb "go.temporal.io/server/api/replication/v1"
 	workflowspb "go.temporal.io/server/api/workflow/v1"
 	"go.temporal.io/server/client/history"
@@ -1081,32 +1081,39 @@ func (e *historyEngineImpl) DescribeMutableState(
 
 	if cacheHit && cacheCtx.(*workflowExecutionContextImpl).mutableState != nil {
 		msb := cacheCtx.(*workflowExecutionContextImpl).mutableState
-		response.CacheMutableState, err = e.toMutableStateJSON(msb)
-		if err != nil {
-			return nil, err
-		}
+		wms := msb.CopyToPersistence()
+		response.CacheMutableState = workflowMutableStateToJSON(wms)
 	}
 
 	msb, err := dbCtx.loadWorkflowExecution()
 	if err != nil {
-		return nil, err
+		response.DatabaseMutableState = err.Error()
+		return response, nil
+	}
+	wms := msb.CopyToPersistence()
+	response.DatabaseMutableState = workflowMutableStateToJSON(wms)
+
+	currentBranchToken := wms.ExecutionInfo.EventBranchToken
+	if wms.VersionHistories != nil {
+		// if VersionHistories is set, then all branch infos are stored in VersionHistories
+		currentVersionHistory, err := wms.VersionHistories.GetCurrentVersionHistory()
+		if err != nil {
+			response.TreeId = err.Error()
+			return response, nil
+		}
+		currentBranchToken = currentVersionHistory.GetBranchToken()
 	}
-	response.DatabaseMutableState, err = e.toMutableStateJSON(msb)
+	branchInfo := persistenceblobs.HistoryBranch{}
+	err = branchInfo.Unmarshal(currentBranchToken)
 	if err != nil {
-		return nil, err
+		response.TreeId = err.Error()
+		return response, nil
 	}
 
-	return response, nil
-}
-
-func (e *historyEngineImpl) toMutableStateJSON(msb mutableState) (string, error) {
-	ms := msb.CopyToPersistence()
+	response.TreeId = branchInfo.TreeId
+	response.BranchId = branchInfo.BranchId
 
-	jsonBytes, err := json.Marshal(ms)
-	if err != nil {
-		return "", err
-	}
-	return string(jsonBytes), nil
+	return response, nil
 }
 
 // ResetStickyTaskQueue reset the volatile information in mutable state of a given workflow.

service/history/workflow_mutable_state_serializer.go

@@ -0,0 +1,163 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package history
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/gogo/protobuf/jsonpb"
+	"github.com/gogo/protobuf/proto"
+
+	"go.temporal.io/server/common/persistence"
+)
+
+func workflowMutableStateToJSON(wms *persistence.WorkflowMutableState) string {
+	// TODO: call @alexshtin before you dig!
+	// This is a _temporarry_ workaround to make JSON serialization work.
+	// This code will be removed soon and MutableState should be serialized to proto.
+
+	jsonMarshal := func(v interface{}) string {
+		if pb, ok := v.(proto.Message); ok {
+			m := jsonpb.Marshaler{}
+			var buf bytes.Buffer
+			err := m.Marshal(&buf, pb)
+			if err != nil {
+				return fmt.Sprintf(`{"error": "%s"}`, err.Error())
+			}
+			return string(buf.Bytes())
+		}
+
+		b, err := json.Marshal(v)
+		if err != nil {
+			return fmt.Sprintf(`{"error": "%s"}`, err.Error())
+		}
+		return string(b)
+	}
+
+	var i int
+	var sb strings.Builder
+	sb.WriteString("{")
+
+	// ActivityInfos
+	sb.WriteString(`"ActivityInfos":`)
+	sb.WriteString("{")
+	i = 1
+	for key, item := range wms.ActivityInfos {
+		sb.WriteString(fmt.Sprintf(`"%d":%s`, key, jsonMarshal(item)))
+		if i != len(wms.ActivityInfos) {
+			sb.WriteString(",")
+		}
+		i++
+	}
+	sb.WriteString("},")
+
+	// TimerInfos
+	sb.WriteString(`"TimerInfos":`)
+	sb.WriteString("{")
+	i = 1
+	for key, item := range wms.TimerInfos {
+		sb.WriteString(fmt.Sprintf(`"%s":%s`, key, jsonMarshal(item)))
+		if i != len(wms.TimerInfos) {
+			sb.WriteString(",")
+		}
+		i++
+	}
+	sb.WriteString("},")
+
+	// ChildExecutionInfos
+	sb.WriteString(`"ChildExecutionInfos":`)
+	sb.WriteString("{")
+	i = 1
+	for key, item := range wms.ChildExecutionInfos {
+		sb.WriteString(fmt.Sprintf(`"%d":%s`, key, jsonMarshal(item)))
+		if i != len(wms.ChildExecutionInfos) {
+			sb.WriteString(",")
+		}
+		i++
+	}
+	sb.WriteString("},")
+
+	// RequestCancelInfos
+	sb.WriteString(`"RequestCancelInfos":`)
+	sb.WriteString("{")
+	i = 1
+	for key, item := range wms.RequestCancelInfos {
+		sb.WriteString(fmt.Sprintf(`"%d":%s`, key, jsonMarshal(item)))
+		if i != len(wms.RequestCancelInfos) {
+			sb.WriteString(",")
+		}
+		i++
+	}
+	sb.WriteString("},")
+
+	// SignalInfos
+	sb.WriteString(`"SignalInfos":`)
+	sb.WriteString("{")
+	i = 1
+	for key, item := range wms.SignalInfos {
+		sb.WriteString(fmt.Sprintf(`"%d":%s`, key, jsonMarshal(item)))
+		if i != len(wms.SignalInfos) {
+			sb.WriteString(",")
+		}
+		i++
+	}
+	sb.WriteString("},")
+
+	// SignalRequestedIDs
+	sb.WriteString(fmt.Sprintf(`"SignalRequestedIds":%s,`, jsonMarshal(wms.SignalRequestedIDs)))
+
+	// ExecutionInfo
+	sb.WriteString(fmt.Sprintf(`"ExecutionInfo":%s,`, jsonMarshal(wms.ExecutionInfo)))
+
+	// ExecutionStats
+	sb.WriteString(fmt.Sprintf(`"ExecutionStats":%s,`, jsonMarshal(wms.ExecutionStats)))
+
+	// BufferedEvents
+	sb.WriteString(`"BufferedEvents":`)
+	sb.WriteString("[")
+	i = 1
+	for _, item := range wms.BufferedEvents {
+		sb.WriteString(jsonMarshal(item))
+		if i != len(wms.BufferedEvents) {
+			sb.WriteString(",")
+		}
+		i++
+	}
+	sb.WriteString("],")
+
+	// VersionHistories
+	sb.WriteString(fmt.Sprintf(`"VersionHistories":%s,`, jsonMarshal(wms.VersionHistories)))
+
+	// Checksum
+	sb.WriteString(fmt.Sprintf(`"Checksum":%s`, jsonMarshal(wms.Checksum)))
+
+	// Close root object
+	sb.WriteString("}")
+
+	return sb.String()
+}