Support trigger continueAsNew API (#385)

service/api/handler.go

@@ -144,6 +144,23 @@ func (h *handler) apiV1WorkflowConfigUpdate(c *gin.Context) {
 	return
 }
 
+func (h *handler) apiV1WorkflowTriggerContinueAsNew(c *gin.Context) {
+	var req iwfidl.TriggerContinueAsNewRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		invalidRequestSchema(c)
+		return
+	}
+	h.logger.Debug("received API request", tag.Value(h.toJson(req)))
+
+	errResp := h.svc.ApiV1WorkflowTriggerContinueAsNew(c.Request.Context(), req)
+	if errResp != nil {
+		h.processError(c, errResp)
+		return
+	}
+	c.JSON(http.StatusOK, struct{}{})
+	return
+}
+
 func (h *handler) apiV1WorkflowSearch(c *gin.Context) {
 	var req iwfidl.WorkflowSearchRequest
 	if err := c.ShouldBindJSON(&req); err != nil {

service/api/interfaces.go

@@ -8,20 +8,43 @@ import (
 )
 
 type ApiService interface {
-	ApiV1WorkflowStartPost(ctx context.Context, request iwfidl.WorkflowStartRequest) (*iwfidl.WorkflowStartResponse, *errors.ErrorAndStatus)
-	ApiV1WorkflowWaitForStateCompletion(ctx context.Context, request iwfidl.WorkflowWaitForStateCompletionRequest) (*iwfidl.WorkflowWaitForStateCompletionResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowStartPost(
+		ctx context.Context, request iwfidl.WorkflowStartRequest,
+	) (*iwfidl.WorkflowStartResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowWaitForStateCompletion(
+		ctx context.Context, request iwfidl.WorkflowWaitForStateCompletionRequest,
+	) (*iwfidl.WorkflowWaitForStateCompletionResponse, *errors.ErrorAndStatus)
 	ApiV1WorkflowSignalPost(ctx context.Context, request iwfidl.WorkflowSignalRequest) *errors.ErrorAndStatus
 	ApiV1WorkflowStopPost(ctx context.Context, request iwfidl.WorkflowStopRequest) *errors.ErrorAndStatus
 	ApiV1WorkflowConfigUpdate(ctx context.Context, request iwfidl.WorkflowConfigUpdateRequest) *errors.ErrorAndStatus
-	ApiV1WorkflowGetQueryAttributesPost(ctx context.Context, request iwfidl.WorkflowGetDataObjectsRequest) (*iwfidl.WorkflowGetDataObjectsResponse, *errors.ErrorAndStatus)
-	ApiV1WorkflowGetSearchAttributesPost(ctx context.Context, request iwfidl.WorkflowGetSearchAttributesRequest) (*iwfidl.WorkflowGetSearchAttributesResponse, *errors.ErrorAndStatus)
-	ApiV1WorkflowGetPost(ctx context.Context, request iwfidl.WorkflowGetRequest) (*iwfidl.WorkflowGetResponse, *errors.ErrorAndStatus)
-	ApiV1WorkflowGetWithWaitPost(ctx context.Context, request iwfidl.WorkflowGetRequest) (*iwfidl.WorkflowGetResponse, *errors.ErrorAndStatus)
-	ApiV1WorkflowSearchPost(ctx context.Context, request iwfidl.WorkflowSearchRequest) (*iwfidl.WorkflowSearchResponse, *errors.ErrorAndStatus)
-	ApiV1WorkflowRpcPost(ctx context.Context, request iwfidl.WorkflowRpcRequest) (*iwfidl.WorkflowRpcResponse, *errors.ErrorAndStatus)
-	ApiV1WorkflowResetPost(ctx context.Context, request iwfidl.WorkflowResetRequest) (*iwfidl.WorkflowResetResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowTriggerContinueAsNew(
+		ctx context.Context, req iwfidl.TriggerContinueAsNewRequest,
+	) (retError *errors.ErrorAndStatus)
+	ApiV1WorkflowGetQueryAttributesPost(
+		ctx context.Context, request iwfidl.WorkflowGetDataObjectsRequest,
+	) (*iwfidl.WorkflowGetDataObjectsResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowGetSearchAttributesPost(
+		ctx context.Context, request iwfidl.WorkflowGetSearchAttributesRequest,
+	) (*iwfidl.WorkflowGetSearchAttributesResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowGetPost(
+		ctx context.Context, request iwfidl.WorkflowGetRequest,
+	) (*iwfidl.WorkflowGetResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowGetWithWaitPost(
+		ctx context.Context, request iwfidl.WorkflowGetRequest,
+	) (*iwfidl.WorkflowGetResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowSearchPost(
+		ctx context.Context, request iwfidl.WorkflowSearchRequest,
+	) (*iwfidl.WorkflowSearchResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowRpcPost(
+		ctx context.Context, request iwfidl.WorkflowRpcRequest,
+	) (*iwfidl.WorkflowRpcResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowResetPost(
+		ctx context.Context, request iwfidl.WorkflowResetRequest,
+	) (*iwfidl.WorkflowResetResponse, *errors.ErrorAndStatus)
 	ApiV1WorkflowSkipTimerPost(ctx context.Context, request iwfidl.WorkflowSkipTimerRequest) *errors.ErrorAndStatus
-	ApiV1WorkflowDumpPost(ctx context.Context, request iwfidl.WorkflowDumpRequest) (*iwfidl.WorkflowDumpResponse, *errors.ErrorAndStatus)
+	ApiV1WorkflowDumpPost(
+		ctx context.Context, request iwfidl.WorkflowDumpRequest,
+	) (*iwfidl.WorkflowDumpResponse, *errors.ErrorAndStatus)
 	ApiInfoHealth(ctx context.Context) *iwfidl.HealthInfo
 	Close()
 }

service/api/routers.go

@@ -20,6 +20,7 @@ const WorkflowSkipTimerApiPath = "/api/v1/workflow/timer/skip"
 const WorkflowStopApiPath = "/api/v1/workflow/stop"
 const WorkflowInternalDumpApiPath = "/api/v1/workflow/internal/dump"
 const WorkflowConfigUpdateApiPath = "/api/v1/workflow/config/update"
+const WorkflowTriggerContinueAsNewApiPath = "/api/v1/workflow/triggerContinueAsNew"
 const WorkflowRpcApiPath = "/api/v1/workflow/rpc"
 const InfoHealthCheck = "/info/healthcheck"
 
@@ -43,6 +44,7 @@ func NewService(config config.Config, client uclient.UnifiedClient, logger log.L
 	router.POST(WorkflowSkipTimerApiPath, handler.apiV1WorkflowSkipTimer)
 	router.POST(WorkflowInternalDumpApiPath, handler.apiV1WorkflowInternalDump)
 	router.POST(WorkflowConfigUpdateApiPath, handler.apiV1WorkflowConfigUpdate)
+	router.POST(WorkflowTriggerContinueAsNewApiPath, handler.apiV1WorkflowTriggerContinueAsNew)
 	router.POST(WorkflowRpcApiPath, handler.apiV1WorkflowRpc)
 	router.GET(InfoHealthCheck, handler.infoHealthCheck)
 

service/api/service.go

@@ -213,6 +213,19 @@ func (s *serviceImpl) ApiV1WorkflowConfigUpdate(
 	return nil
 }
 
+func (s *serviceImpl) ApiV1WorkflowTriggerContinueAsNew(
+	ctx context.Context, req iwfidl.TriggerContinueAsNewRequest,
+) (retError *errors.ErrorAndStatus) {
+	defer func() { log.CapturePanic(recover(), s.logger, &retError) }()
+
+	err := s.client.SignalWorkflow(ctx,
+		req.GetWorkflowId(), req.GetWorkflowRunId(), service.TriggerContinueAsNewSignalChannelName, nil)
+	if err != nil {
+		return s.handleError(err, WorkflowTriggerContinueAsNewApiPath, req.GetWorkflowId())
+	}
+	return nil
+}
+
 func (s *serviceImpl) ApiV1WorkflowStopPost(
 	ctx context.Context, req iwfidl.WorkflowStopRequest,
 ) (retError *errors.ErrorAndStatus) {

service/const.go

@@ -40,11 +40,12 @@ const (
 
 	IwfSystemConstPrefix = "__IwfSystem_"
 
-	SkipTimerSignalChannelName       = IwfSystemConstPrefix + "SkipTimerChannel"
-	FailWorkflowSignalChannelName    = IwfSystemConstPrefix + "FailWorkflowChannel"
-	UpdateConfigSignalChannelName    = IwfSystemConstPrefix + "UpdateWorkflowConfig"
-	ExecuteRpcSignalChannelName      = IwfSystemConstPrefix + "ExecuteRpc"
-	StateCompletionSignalChannelName = IwfSystemConstPrefix + "StateCompletion"
+	SkipTimerSignalChannelName            = IwfSystemConstPrefix + "SkipTimerChannel"
+	FailWorkflowSignalChannelName         = IwfSystemConstPrefix + "FailWorkflowChannel"
+	UpdateConfigSignalChannelName         = IwfSystemConstPrefix + "UpdateWorkflowConfig"
+	ExecuteRpcSignalChannelName           = IwfSystemConstPrefix + "ExecuteRpc"
+	StateCompletionSignalChannelName      = IwfSystemConstPrefix + "StateCompletion"
+	TriggerContinueAsNewSignalChannelName = IwfSystemConstPrefix + "TriggerContinueAsNew"
 
 	WorkerUrlMemoKey            = IwfSystemConstPrefix + "WorkerUrl"
 	UseMemoForDataAttributesKey = IwfSystemConstPrefix + "UseMemoForDataAttributes"

service/interpreter/continueAsNewCounter.go

@@ -4,13 +4,16 @@ type ContinueAsNewCounter struct {
 	executedStateApis  int32
 	signalsReceived    int32
 	syncUpdateReceived int32
+	triggeredByAPI     bool
 
 	configer *WorkflowConfiger
 	rootCtx  UnifiedContext
 	provider WorkflowProvider
 }
 
-func NewContinueAsCounter(configer *WorkflowConfiger, rootCtx UnifiedContext, provider WorkflowProvider) *ContinueAsNewCounter {
+func NewContinueAsCounter(
+	configer *WorkflowConfiger, rootCtx UnifiedContext, provider WorkflowProvider,
+) *ContinueAsNewCounter {
 	return &ContinueAsNewCounter{
 		configer: configer,
 
@@ -35,6 +38,10 @@ func (c *ContinueAsNewCounter) IncSyncUpdateReceived() {
 }
 
 func (c *ContinueAsNewCounter) IsThresholdMet() bool {
+	if c.triggeredByAPI {
+		return true
+	}
+
 	// Note: when threshold == 0, it means unlimited
 
 	config := c.configer.Get()
@@ -45,3 +52,7 @@ func (c *ContinueAsNewCounter) IsThresholdMet() bool {
 
 	return totalOperations >= config.GetContinueAsNewThreshold()
 }
+
+func (c *ContinueAsNewCounter) TriggerByAPI() {
+	c.triggeredByAPI = true
+}

service/interpreter/signalReceiver.go

@@ -20,9 +20,13 @@ type SignalReceiver struct {
 	persistenceManager         *PersistenceManager
 }
 
-func NewSignalReceiver(ctx UnifiedContext, provider WorkflowProvider, interStateChannel *InterStateChannel, stateRequestQueue *StateRequestQueue,
-	persistenceManager *PersistenceManager, tp *TimerProcessor, continueAsNewCounter *ContinueAsNewCounter, workflowConfiger *WorkflowConfiger,
-	initReceivedSignals map[string][]*iwfidl.EncodedObject) *SignalReceiver {
+func NewSignalReceiver(
+	ctx UnifiedContext, provider WorkflowProvider, interStateChannel *InterStateChannel,
+	stateRequestQueue *StateRequestQueue,
+	persistenceManager *PersistenceManager, tp *TimerProcessor, continueAsNewCounter *ContinueAsNewCounter,
+	workflowConfiger *WorkflowConfiger,
+	initReceivedSignals map[string][]*iwfidl.EncodedObject,
+) *SignalReceiver {
 	if initReceivedSignals == nil {
 		initReceivedSignals = map[string][]*iwfidl.EncodedObject{}
 	}
@@ -110,6 +114,26 @@ func NewSignalReceiver(ctx UnifiedContext, provider WorkflowProvider, interState
 		}
 	})
 
+	provider.GoNamed(ctx, "trigger-continue-as-new-handler", func(ctx UnifiedContext) {
+		// NOTE: unlike other signal channels, this one doesn't need to drain during continueAsNew
+		// because if there is a continueAsNew, this signal is not needed anymore
+		ch := provider.GetSignalChannel(ctx, service.TriggerContinueAsNewSignalChannelName)
+
+		received := false
+		err := provider.Await(ctx, func() bool {
+			received = ch.ReceiveAsync(nil)
+			return received || continueAsNewCounter.IsThresholdMet()
+		})
+		if err != nil {
+			return
+		}
+		if received {
+			continueAsNewCounter.TriggerByAPI()
+			return
+		}
+		return
+	})
+
 	provider.GoNamed(ctx, "execute-rpc-signal-handler", func(ctx UnifiedContext) {
 		for {
 			ch := provider.GetSignalChannel(ctx, service.ExecuteRpcSignalChannelName)