Implement reset by stateId and stateExecutionId

README.md

@@ -264,8 +264,8 @@ cadence adm cl asa --search_attr_key IwfWorkflowType --search_attr_type 0
 - [x] AnyCommandCompleted Decider trigger type
 - [x] More workflow start options: IdReusePolicy, cron schedule, retry
 - [x] StateOption: Start/Decide API timeout and retry policy
+- [x] Reset workflow by stateId or stateExecutionId 
 - [ ] More workflow start options: initial search attributes/memo
-- [ ] Reset workflow by stateId
 
 ### 1.2
 - [ ] Decider trigger type: AnyCommandClosed

cmd/server/iwf/iwf.go

@@ -32,9 +32,11 @@ import (
 	apiv1 "github.com/uber/cadence-idl/go/proto/api/v1"
 	"github.com/urfave/cli"
 	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/converter"
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	cclient "go.uber.org/cadence/client"
 	"go.uber.org/cadence/compatibility"
+	"go.uber.org/cadence/encoded"
 	"go.uber.org/yarpc"
 	"go.uber.org/yarpc/transport/grpc"
 	"log"
@@ -99,7 +101,7 @@ func start(c *cli.Context) {
 		if err != nil {
 			log.Fatalf("Unable to connect to Temporal because of error %v", err)
 		}
-		unifiedClient = temporalapi.NewTemporalClient(temporalClient, config.Backend.Temporal.Namespace)
+		unifiedClient = temporalapi.NewTemporalClient(temporalClient, config.Backend.Temporal.Namespace, converter.GetDefaultDataConverter())
 
 		for _, svcName := range services {
 			go launchTemporalService(svcName, config, unifiedClient, temporalClient)
@@ -121,7 +123,7 @@ func start(c *cli.Context) {
 		if err != nil {
 			log.Fatalf("Unable to connect to Cadence because of error %v", err)
 		}
-		unifiedClient = cadenceapi.NewCadenceClient(domain, cadenceClient, serviceClient, closeFunc)
+		unifiedClient = cadenceapi.NewCadenceClient(domain, cadenceClient, serviceClient, encoded.GetDefaultDataConverter(), closeFunc)
 
 		for _, svcName := range services {
 			go launchCadenceService(svcName, config, unifiedClient, serviceClient, domain, closeFunc)

gen/iwfidl/api/openapi.yaml

@@ -709,10 +709,12 @@ components:
         reason: reason
         historyEventId: 0
         historyEventTime: historyEventTime
+        stateId: stateId
         resetType: HISTORY_EVENT_ID
         skipSignalReapply: true
         workflowRunId: workflowRunId
         workflowId: workflowId
+        stateExecutionId: stateExecutionId
       properties:
         workflowId:
           type: string
@@ -723,13 +725,19 @@ components:
           - HISTORY_EVENT_ID
           - BEGINNING
           - HISTORY_EVENT_TIME
+          - STATE_ID
+          - STATE_EXECUTION_ID
           type: string
         historyEventId:
           type: integer
         reason:
           type: string
         historyEventTime:
           type: string
+        stateId:
+          type: string
+        stateExecutionId:
+          type: string
         skipSignalReapply:
           type: boolean
       required:

gen/iwfidl/docs/WorkflowResetRequest.md

@@ -10,6 +10,8 @@ Name | Type | Description | Notes
 **HistoryEventId** | Pointer to **int32** |  | [optional] 
 **Reason** | Pointer to **string** |  | [optional] 
 **HistoryEventTime** | Pointer to **string** |  | [optional] 
+**StateId** | Pointer to **string** |  | [optional] 
+**StateExecutionId** | Pointer to **string** |  | [optional] 
 **SkipSignalReapply** | Pointer to **bool** |  | [optional] 
 
 ## Methods
@@ -171,6 +173,56 @@ SetHistoryEventTime sets HistoryEventTime field to given value.
 
 HasHistoryEventTime returns a boolean if a field has been set.
 
+### GetStateId
+
+`func (o *WorkflowResetRequest) GetStateId() string`
+
+GetStateId returns the StateId field if non-nil, zero value otherwise.
+
+### GetStateIdOk
+
+`func (o *WorkflowResetRequest) GetStateIdOk() (*string, bool)`
+
+GetStateIdOk returns a tuple with the StateId field if it's non-nil, zero value otherwise
+and a boolean to check if the value has been set.
+
+### SetStateId
+
+`func (o *WorkflowResetRequest) SetStateId(v string)`
+
+SetStateId sets StateId field to given value.
+
+### HasStateId
+
+`func (o *WorkflowResetRequest) HasStateId() bool`
+
+HasStateId returns a boolean if a field has been set.
+
+### GetStateExecutionId
+
+`func (o *WorkflowResetRequest) GetStateExecutionId() string`
+
+GetStateExecutionId returns the StateExecutionId field if non-nil, zero value otherwise.
+
+### GetStateExecutionIdOk
+
+`func (o *WorkflowResetRequest) GetStateExecutionIdOk() (*string, bool)`
+
+GetStateExecutionIdOk returns a tuple with the StateExecutionId field if it's non-nil, zero value otherwise
+and a boolean to check if the value has been set.
+
+### SetStateExecutionId
+
+`func (o *WorkflowResetRequest) SetStateExecutionId(v string)`
+
+SetStateExecutionId sets StateExecutionId field to given value.
+
+### HasStateExecutionId
+
+`func (o *WorkflowResetRequest) HasStateExecutionId() bool`
+
+HasStateExecutionId returns a boolean if a field has been set.
+
 ### GetSkipSignalReapply
 
 `func (o *WorkflowResetRequest) GetSkipSignalReapply() bool`

integ/util.go

@@ -11,6 +11,8 @@ import (
 	"github.com/indeedeng/iwf/service/interpreter/cadence"
 	"github.com/indeedeng/iwf/service/interpreter/temporal"
 	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/converter"
+	"go.uber.org/cadence/encoded"
 	"log"
 	"net/http"
 )
@@ -47,7 +49,7 @@ func startWorkflowWorker(handler common.WorkflowHandler) (closeFunc func()) {
 func startIwfService(backendType service.BackendType) (closeFunc func()) {
 	if backendType == service.BackendTypeTemporal {
 		temporalClient := createTemporalClient()
-		iwfService := api.NewService(temporalapi.NewTemporalClient(temporalClient, testNamespace))
+		iwfService := api.NewService(temporalapi.NewTemporalClient(temporalClient, testNamespace, converter.GetDefaultDataConverter()))
 		iwfServer := &http.Server{
 			Addr:    ":" + testIwfServerPort,
 			Handler: iwfService,
@@ -73,7 +75,7 @@ func startIwfService(backendType service.BackendType) (closeFunc func()) {
 
 		cadenceClient, err := iwf.BuildCadenceClient(serviceClient, iwf.DefaultCadenceDomain)
 
-		iwfService := api.NewService(cadenceapi.NewCadenceClient(iwf.DefaultCadenceDomain, cadenceClient, serviceClient, closeFunc))
+		iwfService := api.NewService(cadenceapi.NewCadenceClient(iwf.DefaultCadenceDomain, cadenceClient, serviceClient, encoded.GetDefaultDataConverter(), closeFunc))
 		iwfServer := &http.Server{
 			Addr:    ":" + testIwfServerPort,
 			Handler: iwfService,

script/http/local/home.http

@@ -15,14 +15,16 @@ Content-Type: application/json
   "stateOptions": {}
 }
 
-### test reset API NOTE: try different resetType: BEGINNING, HISTORY_EVENT_ID, HISTORY_EVENT_TIME
+### test reset API NOTE: try different resetType: BEGINNING, HISTORY_EVENT_ID, HISTORY_EVENT_TIME, STATE_ID, STATE_EXECUTION_ID
 POST http://localhost:8801/api/v1/workflow/reset
 Content-Type: application/json
 
 {
   "workflowId": "timer1671219152",
   "workflowRunId": "fae51e69-7e41-4658-89aa-663463951dc7",
-  "resetType": "HISTORY_EVENT_TIME",
+  "resetType": "STATE_EXECUTION_ID",
   "historyEventId":15,
-  "historyEventTime": "150m"
+  "historyEventTime": "150m",
+  "stateId": "S2",
+  "stateExecutionId": "S2-1"
 }

service/api/cadence/client.go

@@ -13,21 +13,24 @@ import (
 	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/client"
+	"go.uber.org/cadence/encoded"
 )
 
 type cadenceClient struct {
 	domain        string
 	cClient       client.Client
 	closeFunc     func()
 	serviceClient workflowserviceclient.Interface
+	converter     encoded.DataConverter
 }
 
-func NewCadenceClient(domain string, cClient client.Client, serviceClient workflowserviceclient.Interface, closeFunc func()) api.UnifiedClient {
+func NewCadenceClient(domain string, cClient client.Client, serviceClient workflowserviceclient.Interface, converter encoded.DataConverter, closeFunc func()) api.UnifiedClient {
 	return &cadenceClient{
 		domain:        domain,
 		cClient:       cClient,
 		closeFunc:     closeFunc,
 		serviceClient: serviceClient,
+		converter:     converter,
 	}
 }
 
@@ -222,7 +225,7 @@ func (t *cadenceClient) ResetWorkflow(ctx context.Context, request iwfidl.Workfl
 
 	resetType := service.ResetType(request.GetResetType())
 	resetBaseRunID, decisionFinishID, err := getResetIDsByType(ctx, resetType, t.domain, request.GetWorkflowId(),
-		reqRunId, t.serviceClient, request.GetHistoryEventId(), request.GetHistoryEventTime())
+		reqRunId, t.serviceClient, t.converter, request.GetHistoryEventId(), request.GetHistoryEventTime(), request.GetStateId(), request.GetStateExecutionId())
 
 	if err != nil {
 		return "", err