Improve waitForStateCompletion API reliability for signalWithStart order

service/interpreter/activityImpl.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/indeedeng/iwf/gen/iwfidl"
 	"github.com/indeedeng/iwf/service"
-	uclient "github.com/indeedeng/iwf/service/client"
 	"github.com/indeedeng/iwf/service/common/compatibility"
 	"github.com/indeedeng/iwf/service/common/config"
 	"github.com/indeedeng/iwf/service/common/rpc"
@@ -23,7 +22,9 @@ func StateStart(ctx context.Context, backendType service.BackendType, input serv
 	return StateApiWaitUntil(ctx, backendType, input)
 }
 
-func StateApiWaitUntil(ctx context.Context, backendType service.BackendType, input service.StateStartActivityInput) (*iwfidl.WorkflowStateStartResponse, error) {
+func StateApiWaitUntil(
+	ctx context.Context, backendType service.BackendType, input service.StateStartActivityInput,
+) (*iwfidl.WorkflowStateStartResponse, error) {
 	provider := getActivityProviderByType(backendType)
 	logger := provider.GetLogger(ctx)
 	logger.Info("StateStartActivity", "input", input)
@@ -62,33 +63,18 @@ func StateDecide(
 	backendType service.BackendType,
 	input service.StateDecideActivityInput,
 	shouldSendSignalOnCompletion bool,
-	timeout time.Duration) (*iwfidl.WorkflowStateDecideResponse, error) {
+	timeout time.Duration,
+) (*iwfidl.WorkflowStateDecideResponse, error) {
 	return StateApiExecute(ctx, backendType, input, shouldSendSignalOnCompletion, timeout)
 }
 
 func StateApiExecute(
 	ctx context.Context,
 	backendType service.BackendType,
 	input service.StateDecideActivityInput,
-	shouldSendSignalOnCompletion bool,
-	timeout time.Duration) (*iwfidl.WorkflowStateDecideResponse, error) {
-	defer func() {
-		if shouldSendSignalOnCompletion {
-			unifiedCleint := env.GetUnifiedClient()
-			err := unifiedCleint.SignalWithStartWaitForStateCompletionWorkflow(
-				ctx, uclient.StartWorkflowOptions{
-					ID:                       service.IwfSystemConstPrefix + input.Request.Context.WorkflowId + "_" + *input.Request.Context.StateExecutionId,
-					TaskQueue:                env.GetTaskQueue(),
-					WorkflowExecutionTimeout: 600 * time.Second, // TODO: make it configurable
-				},
-				iwfidl.StateCompletionOutput{
-					CompletedStateExecutionId: *input.Request.Context.StateExecutionId,
-				})
-			if err != nil {
-				getActivityProviderByType(backendType).GetLogger(ctx).Error("failed to signal on completion", "err", err)
-			}
-		}
-	}()
+	shouldSendSignalOnCompletion bool, // no used anymore, keep for compatibility
+	timeout time.Duration, // no used anymore, keep for compatibility
+) (*iwfidl.WorkflowStateDecideResponse, error) {
 	provider := getActivityProviderByType(backendType)
 	logger := provider.GetLogger(ctx)
 	logger.Info("StateDecideActivity", "input", input)

service/interpreter/cadence/workflowProvider.go

@@ -91,7 +91,9 @@ func (w *workflowProvider) SetQueryHandler(ctx interpreter.UnifiedContext, query
 	return workflow.SetQueryHandler(wfCtx, queryType, handler)
 }
 
-func (w *workflowProvider) SetRpcUpdateHandler(ctx interpreter.UnifiedContext, updateType string, validator interpreter.UnifiedRpcValidator, handler interpreter.UnifiedRpcHandler) error {
+func (w *workflowProvider) SetRpcUpdateHandler(
+	ctx interpreter.UnifiedContext, updateType string, validator interpreter.UnifiedRpcValidator, handler interpreter.UnifiedRpcHandler,
+) error {
 	// NOTE: this feature is not available in Cadence
 	return nil
 }
@@ -192,6 +194,14 @@ func (w *workflowProvider) Now(ctx interpreter.UnifiedContext) time.Time {
 	return workflow.Now(wfCtx)
 }
 
+func (w *workflowProvider) IsReplaying(ctx interpreter.UnifiedContext) bool {
+	wfCtx, ok := ctx.GetContext().(workflow.Context)
+	if !ok {
+		panic("cannot convert to temporal workflow context")
+	}
+	return workflow.IsReplaying(wfCtx)
+}
+
 func (w *workflowProvider) Sleep(ctx interpreter.UnifiedContext, d time.Duration) (err error) {
 	wfCtx, ok := ctx.GetContext().(workflow.Context)
 	if !ok {

service/interpreter/interfaces.go

@@ -96,6 +96,7 @@ type WorkflowProvider interface {
 	WithActivityOptions(ctx UnifiedContext, options ActivityOptions) UnifiedContext
 	ExecuteActivity(ctx UnifiedContext, activity interface{}, args ...interface{}) (future Future)
 	Now(ctx UnifiedContext) time.Time
+	IsReplaying(ctx UnifiedContext) bool
 	Sleep(ctx UnifiedContext, d time.Duration) (err error)
 	NewTimer(ctx UnifiedContext, d time.Duration) Future
 	GetSignalChannel(ctx UnifiedContext, signalName string) (receiveChannel ReceiveChannel)

service/interpreter/temporal/workflowProvider.go

@@ -113,7 +113,9 @@ func (w *workflowProvider) SetQueryHandler(ctx interpreter.UnifiedContext, query
 	return workflow.SetQueryHandler(wfCtx, queryType, handler)
 }
 
-func (w *workflowProvider) SetRpcUpdateHandler(ctx interpreter.UnifiedContext, updateType string, validator interpreter.UnifiedRpcValidator, handler interpreter.UnifiedRpcHandler) error {
+func (w *workflowProvider) SetRpcUpdateHandler(
+	ctx interpreter.UnifiedContext, updateType string, validator interpreter.UnifiedRpcValidator, handler interpreter.UnifiedRpcHandler,
+) error {
 	wfCtx, ok := ctx.GetContext().(workflow.Context)
 	if !ok {
 		panic("cannot convert to temporal workflow context")
@@ -241,6 +243,14 @@ func (w *workflowProvider) Sleep(ctx interpreter.UnifiedContext, d time.Duration
 	return workflow.Sleep(wfCtx, d)
 }
 
+func (w *workflowProvider) IsReplaying(ctx interpreter.UnifiedContext) bool {
+	wfCtx, ok := ctx.GetContext().(workflow.Context)
+	if !ok {
+		panic("cannot convert to temporal workflow context")
+	}
+	return workflow.IsReplaying(wfCtx)
+}
+
 func (w *workflowProvider) GetVersion(ctx interpreter.UnifiedContext, changeID string, minSupported, maxSupported int) int {
 	wfCtx, ok := ctx.GetContext().(workflow.Context)
 	if !ok {

service/interpreter/workflowImpl.go

@@ -1,7 +1,10 @@
 package interpreter
 
 import (
+	"context"
 	"fmt"
+	uclient "github.com/indeedeng/iwf/service/client"
+	"github.com/indeedeng/iwf/service/interpreter/env"
 	"time"
 
 	"github.com/indeedeng/iwf/service/common/compatibility"
@@ -732,6 +735,25 @@ func executeStateDecide(
 		},
 	}, shouldSendSignalOnCompletion, workflowTimeout).Get(ctx, &decideResponse)
 	persistenceManager.UnlockPersistence(saLoadingPolicy, doLoadingPolicy)
+	if err == nil && shouldSendSignalOnCompletion && provider.IsReplaying(ctx) {
+		// NOTE: here uses IsReplaying to signalWithStart, to save an activity for this operation
+		// this is not a problem because the signalWithStart will be very fast and highly available
+		unifiedClient := env.GetUnifiedClient()
+		err := unifiedClient.SignalWithStartWaitForStateCompletionWorkflow(
+			context.Background(),
+			uclient.StartWorkflowOptions{
+				ID:                       service.IwfSystemConstPrefix + executionContext.WorkflowId + "_" + *executionContext.StateExecutionId,
+				TaskQueue:                env.GetTaskQueue(),
+				WorkflowExecutionTimeout: workflowTimeout,
+			},
+			iwfidl.StateCompletionOutput{
+				CompletedStateExecutionId: *executionContext.StateExecutionId,
+			})
+		if err != nil {
+			// for any reasons this fail, just panic and the workflow task will retry
+			panic(fmt.Sprintf("failed to signal on completion %w", err))
+		}
+	}
 	if err != nil {
 		if shouldProceedOnExecuteApiError(state) {
 			return nil, service.ExecuteApiFailedAndProceed, nil