sync thrift files, add RunWorkflow API which starts a workflow and ea… (#328)

* sync thrift files, add ExecuteWorkflow API which starts a workflow and wait until the workflow finishes, return the result, similar to ExecuteChildWorkflow.

* change the signature of GetWorkflowHistory from returning whole history to history event iterator

* expose WorkflowIDReusePolicy along with values

Author: wxing1292

.gen/go/shared/idl.go

@@ -46,6 +46,29 @@ type (
 	// DomainClient is the client for managing operations on the domain.
 	// CLI, tools, ... can use this layer to manager operations on domain.
 	DomainClient = internal.DomainClient
+
+	// HistoryEventIterator is a iterator which can return history events
+	HistoryEventIterator = internal.HistoryEventIterator
+
+	// WorkflowRun represents a started non child workflow
+	WorkflowRun = internal.WorkflowRun
+
+	// WorkflowIDReusePolicy defines workflow ID reuse behavior.
+	WorkflowIDReusePolicy = internal.WorkflowIDReusePolicy
+)
+
+const (
+	// WorkflowIDReusePolicyAllowDuplicateFailedOnly allow start a workflow execution
+	// when workflow not running, and the last execution close state is in
+	// [terminated, cancelled, timeouted, failed].
+	WorkflowIDReusePolicyAllowDuplicateFailedOnly WorkflowIDReusePolicy = internal.WorkflowIDReusePolicyAllowDuplicateFailedOnly
+
+	// WorkflowIDReusePolicyAllowDuplicate allow start a workflow execution using
+	// the same workflow ID,when workflow not running.
+	WorkflowIDReusePolicyAllowDuplicate WorkflowIDReusePolicy = internal.WorkflowIDReusePolicyAllowDuplicate
+
+	// WorkflowIDReusePolicyRejectDuplicate do not allow start a workflow execution using the same workflow ID at all
+	WorkflowIDReusePolicyRejectDuplicate WorkflowIDReusePolicy = internal.WorkflowIDReusePolicyRejectDuplicate
 )
 
 // NewClient creates an instance of a workflow client

.gen/go/shared/types.go

@@ -54,6 +54,24 @@ exception QueryFailedError {
   1: required string message
 }
 
+enum WorkflowIdReusePolicy {
+  /*
+   * allow start a workflow execution using the same workflow ID,
+   * when workflow not running, and the last execution close state is in
+   * [terminated, cancelled, timeouted, failed].
+   */
+  AllowDuplicateFailedOnly,
+  /*
+   * allow start a workflow execution using the same workflow ID,
+   * when workflow not running.
+   */
+  AllowDuplicate,
+  /*
+   * do not allow start a workflow execution using the same workflow ID at all
+   */
+  RejectDuplicate,
+}
+
 enum DomainStatus {
   REGISTERED,
   DEPRECATED,
@@ -167,6 +185,17 @@ enum QueryTaskCompletedType {
   FAILED,
 }
 
+enum PendingActivityState {
+  SCHEDULED,
+  STARTED,
+  CANCEL_REQUESTED,
+}
+
+enum HistoryEventFilterType {
+  ALL_EVENT,
+  CLOSE_EVENT,
+}
+
 struct WorkflowType {
   10: optional string name
 }
@@ -693,6 +722,7 @@ struct StartWorkflowExecutionRequest {
   70: optional i32 taskStartToCloseTimeoutSeconds
   80: optional string identity
   90: optional string requestId
+  100: optional WorkflowIdReusePolicy workflowIdReusePolicy
 }
 
 struct StartWorkflowExecutionResponse {
@@ -828,6 +858,7 @@ struct GetWorkflowExecutionHistoryRequest {
   30: optional i32 maximumPageSize
   40: optional binary nextPageToken
   50: optional bool waitForNewEvent
+  60: optional HistoryEventFilterType HistoryEventFilterType
 }
 
 struct GetWorkflowExecutionHistoryResponse {
@@ -907,7 +938,16 @@ struct DescribeWorkflowExecutionRequest {
   20: optional WorkflowExecution execution
 }
 
+struct PendingActivityInfo {
+  10: optional string activityID
+  20: optional ActivityType activityType
+  30: optional PendingActivityState state
+  40: optional binary heartbeatDetails
+  50: optional i64 (js.type = "Long") lastHeartbeatTimestamp
+}
+
 struct DescribeWorkflowExecutionResponse {
   10: optional WorkflowExecutionConfiguration executionConfiguration
   20: optional WorkflowExecutionInfo workflowExecutionInfo
+  30: optional list<PendingActivityInfo> pendingActivities
 }

client/client.go

@@ -22,6 +22,7 @@ package internal
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"go.uber.org/cadence/encoded"
@@ -43,15 +44,42 @@ type (
 		// StartWorkflow starts a workflow execution
 		// The user can use this to start using a function or workflow type name.
 		// Either by
-		//     StartWorkflow(ctx, options, "workflowTypeName", input)
+		//     StartWorkflow(ctx, options, "workflowTypeName", arg1, arg2, arg3)
 		//     or
 		//     StartWorkflow(ctx, options, workflowExecuteFn, arg1, arg2, arg3)
 		// The errors it can return:
-		//	- EntityNotExistsError
+		//	- EntityNotExistsError, if domain does not exists
 		//	- BadRequestError
 		//	- WorkflowExecutionAlreadyStartedError
+		//	- InternalServiceError
 		StartWorkflow(ctx context.Context, options StartWorkflowOptions, workflow interface{}, args ...interface{}) (*WorkflowExecution, error)
 
+		// ExecuteWorkflow starts a workflow execution and return a WorkflowRun instance and error
+		// The user can use this to start using a function or workflow type name.
+		// Either by
+		//     ExecuteWorkflow(ctx, options, "workflowTypeName", arg1, arg2, arg3)
+		//     or
+		//     ExecuteWorkflow(ctx, options, workflowExecuteFn, arg1, arg2, arg3)
+		// The errors it can return:
+		//	- EntityNotExistsError, if domain does not exists
+		//	- BadRequestError
+		//	- WorkflowExecutionAlreadyStartedError
+		//	- InternalServiceError
+		//
+		// WorkflowRun has 2 methods:
+		//  - GetRunID() string: which return the first started workflow run ID (please see below)
+		//  - Get(ctx context.Context, valuePtr interface{}) error: which will fill the workflow
+		//    execution result to valuePtr, if workflow execution is a success, or return corresponding
+		//    error. This is a blocking API.
+		// NOTE: if the started workflow return ContinueAsNewError during the workflow execution, the
+		// return result of GetRunID() will be the started workflow run ID, not the new run ID caused by ContinueAsNewError,
+		// however, Get(ctx context.Context, valuePtr interface{}) will return result from the run which did not return ContinueAsNewError.
+		// Say ExecuteWorkflow started a workflow, in its first run, has run ID "run ID 1", and returned ContinueAsNewError,
+		// the second run has run ID "run ID 2" and return some result other than ContinueAsNewError:
+		// GetRunID() will always return "run ID 1" and  Get(ctx context.Context, valuePtr interface{}) will return the result of second run.
+		// NOTE: DO NOT USE THIS API INSIDE A WORKFLOW, USE workflow.ExecuteChildWorkflow instead
+		ExecuteWorkflow(ctx context.Context, options StartWorkflowOptions, workflow interface{}, args ...interface{}) (WorkflowRun, error)
+
 		// SignalWorkflow sends a signals to a workflow in execution
 		// - workflow ID of the workflow.
 		// - runID can be default(empty string). if empty string then it will pick the running execution of that workflow ID.
@@ -80,14 +108,25 @@ type (
 		//	- InternalServiceError
 		TerminateWorkflow(ctx context.Context, workflowID string, runID string, reason string, details []byte) error
 
-		// GetWorkflowHistory gets history of a particular workflow.
+		// GetWorkflowHistory gets history events of a particular workflow
 		// - workflow ID of the workflow.
-		// - runID can be default(empty string). if empty string then it will pick the running execution of that workflow ID.
-		// The errors it can return:
-		//	- EntityNotExistsError
-		//	- BadRequestError
-		//	- InternalServiceError
-		GetWorkflowHistory(ctx context.Context, workflowID string, runID string) (*s.History, error)
+		// - runID can be default(empty string). if empty string then it will pick the last running execution of that workflow ID.
+		// - whether use long poll for tracking new events: when the workflow is running, there can be new events generated during iteration
+		// 	 of HistoryEventIterator, if isLongPoll == true, then iterator will do long poll, tracking new history event, i.e. the iteration
+		//   will not be finished until workflow is finished; if isLongPoll == false, then iterator will only return current history events.
+		// - whether return all history events or just the last event, which contains the workflow execution end result
+		// Example:-
+		//	To iterate all events,
+		//		iter := GetWorkflowHistory(ctx, workflowID, runID, isLongPoll, filterType)
+		//		events := []*shared.HistoryEvent{}
+		//		for iter.HasNext() {
+		//			event, err := iter.Next()
+		//			if err != nil {
+		//				return err
+		//			}
+		//			events = append(events, event)
+		//		}
+		GetWorkflowHistory(ctx context.Context, workflowID string, runID string, isLongPoll bool, filterType s.HistoryEventFilterType) HistoryEventIterator
 
 		// CompleteActivity reports activity completed.
 		// activity Execute method can return acitivity.activity.ErrResultPending to
@@ -147,7 +186,7 @@ type (
 		// run. By default, cadence supports "__stack_trace" as a standard query type, which will return string value
 		// representing the call stack of the target workflow. The target workflow could also setup different query handler
 		// to handle custom query types.
-		// See comments at cadence.SetQueryHandler(ctx Context, queryType string, handler interface{}) for more details
+		// See comments at workflow.SetQueryHandler(ctx Context, queryType string, handler interface{}) for more details
 		// on how to setup query handler within the target workflow.
 		// - workflowID is required.
 		// - runID can be default(empty string). if empty string then it will pick the running execution of that workflow ID.
@@ -196,6 +235,10 @@ type (
 		// The resolution is seconds.
 		// Optional: defaulted to 20 secs.
 		DecisionTaskStartToCloseTimeout time.Duration
+
+		// WorkflowIDReusePolicy - Whether server allow reuse of workflow ID, can be useful
+		// for dedup logic if set to WorkflowIdReusePolicyRejectDuplicate
+		WorkflowIDReusePolicy WorkflowIDReusePolicy
 	}
 
 	// DomainClient is the client for managing operations on the domain.
@@ -226,6 +269,23 @@ type (
 		//	- InternalServiceError
 		Update(ctx context.Context, name string, domainInfo *s.UpdateDomainInfo, domainConfig *s.DomainConfiguration) error
 	}
+
+	// WorkflowIDReusePolicy defines workflow ID reuse behavior.
+	WorkflowIDReusePolicy int
+)
+
+const (
+	// WorkflowIDReusePolicyAllowDuplicateFailedOnly allow start a workflow execution
+	// when workflow not running, and the last execution close state is in
+	// [terminated, cancelled, timeouted, failed].
+	WorkflowIDReusePolicyAllowDuplicateFailedOnly WorkflowIDReusePolicy = iota
+
+	// WorkflowIDReusePolicyAllowDuplicate allow start a workflow execution using
+	// the same workflow ID,when workflow not running.
+	WorkflowIDReusePolicyAllowDuplicate
+
+	// WorkflowIDReusePolicyRejectDuplicate do not allow start a workflow execution using the same workflow ID at all
+	WorkflowIDReusePolicyRejectDuplicate
 )
 
 // NewClient creates an instance of a workflow client
@@ -268,3 +328,18 @@ func NewDomainClient(service workflowserviceclient.Interface, options *ClientOpt
 		identity:        identity,
 	}
 }
+
+func (p WorkflowIDReusePolicy) toThriftPtr() *s.WorkflowIdReusePolicy {
+	var policy s.WorkflowIdReusePolicy
+	switch p {
+	case WorkflowIDReusePolicyAllowDuplicate:
+		policy = s.WorkflowIdReusePolicyAllowDuplicate
+	case WorkflowIDReusePolicyAllowDuplicateFailedOnly:
+		policy = s.WorkflowIdReusePolicyAllowDuplicateFailedOnly
+	case WorkflowIDReusePolicyRejectDuplicate:
+		policy = s.WorkflowIdReusePolicyRejectDuplicate
+	default:
+		panic(fmt.Sprintf("unknown workflow reuse policy %v", p))
+	}
+	return &policy
+}

idl/github.com/uber/cadence/shared.thrift

@@ -55,10 +55,10 @@ Workflow code could handle errors based on different types of error. Below is sa
 _, err := workflow.ExecuteActivity(ctx, MyActivity, ...).Get(nil)
 if err != nil {
 	switch err := err.(type) {
-	case *CustomError:
+	case *workflow.CustomError:
 		// handle activity errors (created via NewCustomError() API)
 		switch err.Reason() {
-		case cadence.CustomErrReasonA: // assume CustomErrReasonA is constant defined by activity implementation
+		case CustomErrReasonA: // assume CustomErrReasonA is constant defined by activity implementation
 			var detailMsg string // assuming activity return error by NewCustomError(CustomErrReasonA, "string details")
 			err.Details(&detailMsg) // extract strong typed details (corresponding to CustomErrReasonA)
 			// handle CustomErrReasonA
@@ -67,13 +67,13 @@ if err != nil {
 		default:
 			// newer version of activity could return new errors that workflow was not aware of.
 		}
-	case *cadence.GenericError:
+	case *workflow.GenericError:
 		// handle generic error (errors created other than using NewCustomError() API)
-	case *cadence.CanceledError:
+	case *workflow.CanceledError:
 		// handle cancellation
-	case *cadence.TimeoutError:
+	case *workflow.TimeoutError:
 		// handle timeout, could check timeout type by err.TimeoutType()
-	case *cadence.PanicError:
+	case *workflow.PanicError:
 		// handle panic
 	}
 }
@@ -106,6 +106,10 @@ type (
 		details []byte
 	}
 
+	// TerminatedError returned when workflow was terminated.
+	TerminatedError struct {
+	}
+
 	// PanicError contains information about panicked workflow/activity.
 	PanicError struct {
 		value      string
@@ -276,3 +280,13 @@ func (e *PanicError) StackTrace() string {
 func (e *ContinueAsNewError) Error() string {
 	return "ContinueAsNew"
 }
+
+// newTerminatedError creates NewTerminatedError instance
+func newTerminatedError() *TerminatedError {
+	return &TerminatedError{}
+}
+
+// Error from error interface
+func (e *TerminatedError) Error() string {
+	return "Terminated"
+}

internal/client.go

@@ -813,7 +813,7 @@ func (weh *workflowExecutionEventHandlerImpl) handleChildWorkflowExecutionTermin
 	if childWorkflow.handled {
 		return nil
 	}
-	err := errors.New("terminated")
+	err := newTerminatedError()
 	childWorkflow.handle(nil, err)
 
 	return nil

internal/error.go

@@ -1244,7 +1244,7 @@ func recordActivityHeartbeat(
 }
 
 // This enables verbose logging in the client library.
-// check Cadence.EnableVerboseLogging()
+// check worker.EnableVerboseLogging()
 func traceLog(fn func()) {
 	if enableVerboseLogging {
 		fn()

internal/internal_event_handlers.go

@@ -83,7 +83,7 @@ type (
 	}
 
 	// Workflow is an interface that any workflow should implement.
-	// Code of a workflow must be deterministic. It must use cadence.Channel, cadence.Selector, and workflow.Go instead of
+	// Code of a workflow must be deterministic. It must use workflow.Channel, workflow.Selector, and workflow.Go instead of
 	// native channels, select and go. It also must not use range operation over map as it is randomized by go runtime.
 	// All time manipulation should use current time returned by GetTime(ctx) method.
 	// Note that workflow.Context is used instead of context.Context to avoid use of raw channels.