Merge pull request #211 from cschleiden/continue-as-new

Implement `ContinueAsNew`

Author: cschleiden

README.md

@@ -423,7 +423,24 @@ func SubWorkflow(ctx workflow.Context, msg string) (int, error) {
 
 Similar to timer cancellation, you can pass a cancelable context to `CreateSubWorkflowInstance` and cancel the sub-workflow that way. Reacting to the cancellation is the same as canceling a workflow via the `Client`. See [Canceling workflows](#canceling-workflows) for more details.
 
+### `ContinueAsNew`
+
+```ContinueAsNew` allows you to restart workflow execution with different inputs. The purpose is to keep the history size small enough to avoid hitting size limits, running out of memory and impacting performance. It works by returning a special `error` from your workflow that contains the new inputs:
+
+```go
+wf := func(ctx workflow.Context, run int) (int, error) {
+	run = run + 1
+	if run < 3 {
+		return run, workflow.ContinueAsNew(ctx, run)
+	}
+
+	return run, nil
+}
+```
+
+Here the workflow is going to be restarted when `workflow.ContinueAsNew` is returned. Internally the new execution starts with a fresh history. It uses the same `InstanceID` but a different `ExecutionID`.
 
+If a sub-workflow is restarted, the caller doesn't notice this, only once it ends without being restarted the caller will get the result and control will be passed back.
 
 ### `select`
 
@@ -741,4 +758,4 @@ This kind of check is understandable for simple changes, but it becomes hard and
 
 ### `ContinueAsNew`
 
-Both Temporal/Cadence and DTFx support `ContinueAsNew`. This essentially re-starts a running workflow as a new workflow with a new event history. This is needed for long running workflows where the history can become very large, negatively affecting performance. This feature is not implemented in go-workflows.
+Both Temporal/Cadence and DTFx support `ContinueAsNew`. This essentially re-starts a running workflow as a new workflow with a new event history. This is needed for long running workflows where the history can become very large, negatively affecting performance. While `WorkflowInstance` supports an `InstanceID` and an `ExecutionID`, this feature is not yet implemented (and might not be).

activitytester/activitytester.go

@@ -14,5 +14,5 @@ func WithActivityTestState(ctx context.Context, activityID, instanceID string, l
 		logger = dlogger.NewDefaultLogger()
 	}
 
-	return activity.WithActivityState(ctx, activity.NewActivityState(activityID, core.NewWorkflowInstance(instanceID), logger))
+	return activity.WithActivityState(ctx, activity.NewActivityState(activityID, core.NewWorkflowInstance(instanceID, ""), logger))
 }

backend/mysql/diagnostics.go

@@ -11,7 +11,7 @@ import (
 
 var _ diag.Backend = (*mysqlBackend)(nil)
 
-func (mb *mysqlBackend) GetWorkflowInstances(ctx context.Context, afterInstanceID string, count int) ([]*diag.WorkflowInstanceRef, error) {
+func (mb *mysqlBackend) GetWorkflowInstances(ctx context.Context, afterInstanceID, afterExecutionID string, count int) ([]*diag.WorkflowInstanceRef, error) {
 	var err error
 	tx, err := mb.db.BeginTx(ctx, nil)
 	if err != nil {
@@ -23,19 +23,20 @@ func (mb *mysqlBackend) GetWorkflowInstances(ctx context.Context, afterInstanceI
 	if afterInstanceID != "" {
 		rows, err = tx.QueryContext(
 			ctx,
-			`SELECT i.instance_id, i.created_at, i.completed_at
+			`SELECT i.instance_id, i.execution_id, i.created_at, i.completed_at
 			FROM instances i
-			INNER JOIN (SELECT instance_id, created_at FROM instances WHERE id = ?) ii
+			INNER JOIN (SELECT instance_id, created_at FROM instances WHERE id = ? AND execution_id = ?) ii
 				ON i.created_at < ii.created_at OR (i.created_at = ii.created_at AND i.instance_id < ii.instance_id)
 			ORDER BY i.created_at DESC, i.instance_id DESC
 			LIMIT ?`,
 			afterInstanceID,
+			afterExecutionID,
 			count,
 		)
 	} else {
 		rows, err = tx.QueryContext(
 			ctx,
-			`SELECT i.instance_id, i.created_at, i.completed_at
+			`SELECT i.instance_id, i.execution_id, i.created_at, i.completed_at
 			FROM instances i
 			ORDER BY i.created_at DESC, i.instance_id DESC
 			LIMIT ?`,
@@ -49,10 +50,10 @@ func (mb *mysqlBackend) GetWorkflowInstances(ctx context.Context, afterInstanceI
 	var instances []*diag.WorkflowInstanceRef
 
 	for rows.Next() {
-		var id string
+		var id, executionID string
 		var createdAt time.Time
 		var completedAt *time.Time
-		err = rows.Scan(&id, &createdAt, &completedAt)
+		err = rows.Scan(&id, &executionID, &createdAt, &completedAt)
 		if err != nil {
 			return nil, err
 		}
@@ -63,7 +64,7 @@ func (mb *mysqlBackend) GetWorkflowInstances(ctx context.Context, afterInstanceI
 		}
 
 		instances = append(instances, &diag.WorkflowInstanceRef{
-			Instance:    core.NewWorkflowInstance(id),
+			Instance:    core.NewWorkflowInstance(id, executionID),
 			CreatedAt:   createdAt,
 			CompletedAt: completedAt,
 			State:       state,
@@ -73,20 +74,22 @@ func (mb *mysqlBackend) GetWorkflowInstances(ctx context.Context, afterInstanceI
 	return instances, nil
 }
 
-func (mb *mysqlBackend) GetWorkflowInstance(ctx context.Context, instanceID string) (*diag.WorkflowInstanceRef, error) {
+func (mb *mysqlBackend) GetWorkflowInstance(ctx context.Context, instance *core.WorkflowInstance) (*diag.WorkflowInstanceRef, error) {
 	tx, err := mb.db.BeginTx(ctx, nil)
 	if err != nil {
 		return nil, err
 	}
 	defer tx.Rollback()
 
-	res := tx.QueryRowContext(ctx, "SELECT instance_id, created_at, completed_at FROM instances WHERE instance_id = ?", instanceID)
+	res := tx.QueryRowContext(
+		ctx,
+		"SELECT instance_id, execution_id, created_at, completed_at FROM instances WHERE instance_id = ? AND execution_id = ?", instance.InstanceID, instance.ExecutionID)
 
-	var id string
+	var id, executionID string
 	var createdAt time.Time
 	var completedAt *time.Time
 
-	err = res.Scan(&id, &createdAt, &completedAt)
+	err = res.Scan(&id, &executionID, &createdAt, &completedAt)
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil, nil
@@ -101,14 +104,14 @@ func (mb *mysqlBackend) GetWorkflowInstance(ctx context.Context, instanceID stri
 	}
 
 	return &diag.WorkflowInstanceRef{
-		Instance:    core.NewWorkflowInstance(id),
+		Instance:    core.NewWorkflowInstance(id, executionID),
 		CreatedAt:   createdAt,
 		CompletedAt: completedAt,
 		State:       state,
 	}, nil
 }
 
-func (mb *mysqlBackend) GetWorkflowTree(ctx context.Context, instanceID string) (*diag.WorkflowInstanceTree, error) {
+func (mb *mysqlBackend) GetWorkflowTree(ctx context.Context, instance *core.WorkflowInstance) (*diag.WorkflowInstanceTree, error) {
 	itb := diag.NewInstanceTreeBuilder(mb)
-	return itb.BuildWorkflowInstanceTree(ctx, instanceID)
+	return itb.BuildWorkflowInstanceTree(ctx, instance)
 }

backend/mysql/events.go

@@ -5,18 +5,19 @@ import (
 	"database/sql"
 	"strings"
 
+	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
 )
 
-func insertPendingEvents(ctx context.Context, tx *sql.Tx, instanceID string, newEvents []*history.Event) error {
-	return insertEvents(ctx, tx, "pending_events", instanceID, newEvents)
+func insertPendingEvents(ctx context.Context, tx *sql.Tx, instance *core.WorkflowInstance, newEvents []*history.Event) error {
+	return insertEvents(ctx, tx, "pending_events", instance, newEvents)
 }
 
-func insertHistoryEvents(ctx context.Context, tx *sql.Tx, instanceID string, historyEvents []*history.Event) error {
-	return insertEvents(ctx, tx, "history", instanceID, historyEvents)
+func insertHistoryEvents(ctx context.Context, tx *sql.Tx, instance *core.WorkflowInstance, historyEvents []*history.Event) error {
+	return insertEvents(ctx, tx, "history", instance, historyEvents)
 }
 
-func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instanceID string, events []*history.Event) error {
+func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instance *core.WorkflowInstance, events []*history.Event) error {
 	const batchSize = 20
 	for batchStart := 0; batchStart < len(events); batchStart += batchSize {
 		batchEnd := batchStart + batchSize
@@ -25,8 +26,9 @@ func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instanceID
 		}
 		batchEvents := events[batchStart:batchEnd]
 
-		query := "INSERT INTO `" + tableName + "` (event_id, sequence_id, instance_id, event_type, timestamp, schedule_event_id, attributes, visible_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)" +
-			strings.Repeat(", (?, ?, ?, ?, ?, ?, ?, ?)", len(batchEvents)-1)
+		query := "INSERT INTO `" + tableName +
+			"` (event_id, sequence_id, instance_id, execution_id, event_type, timestamp, schedule_event_id, attributes, visible_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)" +
+			strings.Repeat(", (?, ?, ?, ?, ?, ?, ?, ?, ?)", len(batchEvents)-1)
 
 		args := make([]interface{}, 0, len(batchEvents)*7)
 
@@ -36,7 +38,9 @@ func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instanceID
 				return err
 			}
 
-			args = append(args, newEvent.ID, newEvent.SequenceID, instanceID, newEvent.Type, newEvent.Timestamp, newEvent.ScheduleEventID, a, newEvent.VisibleAt)
+			args = append(
+				args,
+				newEvent.ID, newEvent.SequenceID, instance.InstanceID, instance.ExecutionID, newEvent.Type, newEvent.Timestamp, newEvent.ScheduleEventID, a, newEvent.VisibleAt)
 		}
 
 		_, err := tx.ExecContext(
@@ -52,11 +56,12 @@ func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instanceID
 	return nil
 }
 
-func removeFutureEvent(ctx context.Context, tx *sql.Tx, instanceID string, scheduleEventID int64) error {
+func removeFutureEvent(ctx context.Context, tx *sql.Tx, instance *core.WorkflowInstance, scheduleEventID int64) error {
 	_, err := tx.ExecContext(
 		ctx,
-		"DELETE FROM `pending_events` WHERE instance_id = ? AND schedule_event_id = ? AND visible_at IS NOT NULL",
-		instanceID,
+		"DELETE FROM `pending_events` WHERE instance_id = ? AND execution_id = ? AND schedule_event_id = ? AND visible_at IS NOT NULL",
+		instance.InstanceID,
+		instance.ExecutionID,
 		scheduleEventID,
 	)