Merge pull request #56 from cschleiden/sub-workflow-cancellation

Support sub-workflow cancellation via context cancellation

Author: cschleiden

.github/workflows/go.yml

@@ -31,4 +31,4 @@ jobs:
     - name: Tests
       run: |
         sudo /etc/init.d/mysql start
-        go test -v ./...
+        go test -race -timeout 30s -count 1 ./...

.vscode/launch.json

@@ -41,11 +41,11 @@
       "program": "${workspaceFolder}/samples/signal/signal.go"
     },
     {
-      "name": "Launch cancel sample",
+      "name": "Launch cancellation sample",
       "type": "go",
       "request": "launch",
       "mode": "debug",
-      "program": "${workspaceFolder}/samples/cancelation/cancelation.go"
+      "program": "${workspaceFolder}/samples/cancellation/cancellation.go"
     },
     {
       "name": "Launch subworkflow sample",

README.md

@@ -237,9 +237,57 @@ wf, err := c.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
 if err != nil {
 ```
 
+### Canceling workflows
+
+Create a `Client` instance then then call `CancelWorkflow` to cancel a workflow. When a workflow is canceled, it's workflow context is canceled. Any subsequent calls to schedule activities or sub-workflows will immediately return an error, skipping their execution. Any activities already running when a workflow is canceled will still run to completion and their result will be available.
+
+Sub-workflows will be canceled if their parent workflow is canceled.
+
+```go
+var c client.Client
+err = c.CancelWorkflowInstance(context.Background(), workflowInstance)
+if err != nil {
+	panic("could not cancel workflow")
+}
+```
+
+#### Perform any cleanup
+
+If you need to run any activities or make calls using `workflow.Context`.
+
+```go
+func Workflow2(ctx workflow.Context, msg string) (string, error) {
+	defer func() {
+		if errors.Is(ctx.Err(), workflow.Canceled) {
+			// Workflow was canceled. Get new context to perform any cleanup activities
+			ctx := workflow.NewDisconnectedContext(ctx)
+
+			// Execute the cleanup activity
+			if err := workflow.ExecuteActivity(ctx, ActivityCleanup).Get(ctx, nil); err != nil {
+				return errors.Wrap(err, "could not perform cleanup")
+			}
+		}
+	}()
+
+	r1, err := workflow.ExecuteActivity[int](ctx, ActivityCancel, 1, 2).Get(ctx)
+	if err != nil {  // <---- Workflow is canceled while this activity is running
+		return errors.Wrap(err, "could not get ActivityCancel result")
+	}
+
+	// r1 will contain the result of ActivityCancel
+	// ⬇ ActivitySkip will be skipped immediately
+	r2, err := workflow.ExecuteActivity(ctx, ActivitySkip, 1, 2).Get(ctx)
+	if err != nil {
+		return errors.Wrap(err, "could not get ActivitySkip result")
+	}
+
+	return "some result", nil
+}
+```
+
 ### Running activities
 
-From a workflow, call `workflow.ExecuteActivity` to execute an activity. The call returns a `Future` you can await to get the result or any error it might return.
+From a workflow, call `workflow.ExecuteActivity` to execute an activity. The call returns a `Future[T]` you can await to get the result or any error it might return.
 
 ```go
 r1, err := workflow.ExecuteActivity[int](ctx, Activity1, 35, 12, nil, "test").Get(ctx)
@@ -250,6 +298,10 @@ if err != nil {
 log.Println(r1)
 ```
 
+#### Canceling activities
+
+Canceling activities is not supported at this time.
+
 ### Timers
 
 You can schedule timers to fire at any point in the future by calling `workflow.ScheduleTimer`. It returns a `Future` you can await to wait for the timer to fire.
@@ -283,44 +335,37 @@ id, _ := workflow.SideEffect[string](ctx, func(ctx workflow.Context) string) {
 
 ### Running sub-workflows
 
-Call `workflow.CreateSubWorkflowInstance` to start a sub-workflow.
+Call `workflow.CreateSubWorkflowInstance` to start a sub-workflow. The returned `Future` will resolve once the sub-workflow has finished.
 
 ```go
 func Workflow1(ctx workflow.Context, msg string) error {
-	wr, err := workflow.CreateSubWorkflowInstance[int](ctx, workflow.SubWorkflowInstanceOptions{}, Workflow2, "some input").Get(ctx, &wr)
+	result, err := workflow.CreateSubWorkflowInstance[int]
+		ctx, workflow.SubWorkflowInstanceOptions{}, SubWorkflow, "some input").Get(ctx)
 	if err != nil {
 		return errors.Wrap(err, "could not get sub workflow result")
 	}
 
-	log.Println("Sub workflow result:", wr)
+	logger.Debug("Sub workflow result:", "result", result)
+
 	return nil
 }
 
-func Workflow2(ctx workflow.Context, msg string) (int, error) {
-	r1, err := workflow.ExecuteActivity[int](ctx, Activity1, 35, 12).Get(ctx, &r1)
+func SubWorkflow(ctx workflow.Context, msg string) (int, error) {
+	r1, err := workflow.ExecuteActivity[int](ctx, Activity1, 35, 12).Get(ctx)
 	if err != nil {
 		return "", errors.Wrap(err, "could not get activity result")
 	}
 
-	log.Println("A1 result:", r1)
-
+	logger.Debug("A1 result:", "r1", r1)
 	return r1, nil
 }
 ```
 
-### Canceling workflows
+#### Canceling sub-workflows
 
-Create a `Client` instance then then call `CancelWorkflow` to cancel a workflow. When a workflow is canceled, it's workflow context is canceled. Any subsequent calls to schedule activities or sub-workflows will immediately return an error, skipping their execution. Activities or sub-workflows already running when a workflow is canceled will still run to completion and their result will be available.
+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.
 
-Sub-workflows will be canceled if their parent workflow is canceled.
 
-```go
-var c client.Client
-err = c.CancelWorkflowInstance(context.Background(), workflowInstance)
-if err != nil {
-	panic("could not cancel workflow")
-}
-```
 
 ### `select`
 
@@ -399,39 +444,6 @@ workflow.Select(
 )
 ```
 
-
-#### Perform any cleanup
-
-```go
-func Workflow2(ctx workflow.Context, msg string) (string, error) {
-	defer func() {
-		if errors.Is(ctx.Err(), workflow.Canceled) {
-			// Workflow was canceled. Get new context to perform any cleanup activities
-			ctx := workflow.NewDisconnectedContext(ctx)
-
-			// Execute the cleanup activity
-			if err := workflow.ExecuteActivity(ctx, ActivityCleanup).Get(ctx, nil); err != nil {
-				return errors.Wrap(err, "could not perform cleanup")
-			}
-		}
-	}()
-
-	r1, err := workflow.ExecuteActivity[int](ctx, ActivityCancel, 1, 2).Get(ctx)
-	if err != nil {  // <---- Workflow is canceled while this activity is running
-		return errors.Wrap(err, "could not get ActivityCancel result")
-	}
-
-	// r1 will contain the result of ActivityCancel
-	// ⬇ ActivitySkip will be skipped immediately
-	r2, err := workflow.ExecuteActivity(ctx, ActivitySkip, 1, 2).Get(ctx)
-	if err != nil {
-		return errors.Wrap(err, "could not get ActivitySkip result")
-	}
-
-	return "some result", nil
-}
-```
-
 ### Unit testing
 
 go-workflows includes support for testing workflows, a simple example using mocked activities:

backend/mysql/mysql.go

@@ -89,7 +89,9 @@ func (b *mysqlBackend) Logger() log.Logger {
 }
 
 func (b *mysqlBackend) CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {
-	tx, err := b.db.BeginTx(ctx, nil)
+	tx, err := b.db.BeginTx(ctx, &sql.TxOptions{
+		Isolation: sql.LevelReadCommitted,
+	})
 	if err != nil {
 		return err
 	}
@@ -98,30 +100,37 @@ func (b *mysqlBackend) CancelWorkflowInstance(ctx context.Context, instance *wor
 	instanceID := instance.InstanceID
 
 	// Cancel workflow instance
-	if err := insertNewEvents(ctx, tx, instanceID, []history.Event{*event}); err != nil {
-		return fmt.Errorf("inserting cancellation event: %w", err)
+	// TODO: Combine this with the event insertion
+	res := tx.QueryRowContext(ctx, "SELECT 1 FROM `instances` WHERE instance_id = ? LIMIT 1", instanceID)
+	if err := res.Scan(nil); err == sql.ErrNoRows {
+		return backend.ErrInstanceNotFound
 	}
 
 	// Recursively, find any sub-workflow instance to cancel
-	for {
-		row := tx.QueryRowContext(ctx, "SELECT instance_id FROM `instances` WHERE parent_instance_id = ? AND completed_at IS NULL LIMIT 1", instanceID)
-
-		var subWorkflowInstanceID string
-		if err := row.Scan(&subWorkflowInstanceID); err != nil {
-			if err == sql.ErrNoRows {
-				// No more sub-workflow instances to cancel
-				break
-			}
+	toCancel := []string{instance.InstanceID}
 
-			return fmt.Errorf("getting workflow instance for cancelling: %w", err)
-		}
+	for len(toCancel) > 0 {
+		toCancelID := toCancel[0]
+		toCancel = toCancel[1:]
 
-		// Cancel sub-workflow instance
-		if err := insertNewEvents(ctx, tx, subWorkflowInstanceID, []history.Event{*event}); err != nil {
+		if err := insertNewEvents(ctx, tx, toCancelID, []history.Event{*event}); err != nil {
 			return fmt.Errorf("inserting cancellation event: %w", err)
 		}
 
-		instanceID = subWorkflowInstanceID
+		rows, err := tx.QueryContext(ctx, "SELECT instance_id FROM `instances` WHERE parent_instance_id = ? AND completed_at IS NULL", toCancelID)
+		defer rows.Close()
+		if err != nil {
+			return fmt.Errorf("finding sub-workflow instances: %w", err)
+		}
+
+		for rows.Next() {
+			var subWorkflowInstanceID string
+			if err := rows.Scan(&subWorkflowInstanceID); err != nil {
+				return fmt.Errorf("geting workflow instance for canceling: %w", err)
+			}
+
+			toCancel = append(toCancel, subWorkflowInstanceID)
+		}
 	}
 
 	return tx.Commit()

backend/mysql/mysql_test.go

@@ -22,14 +22,15 @@ func Test_MysqlBackend(t *testing.T) {
 		t.Skip()
 	}
 
-	dbName := "test_" + strings.Replace(uuid.NewString(), "-", "", -1)
+	var dbName string
 
 	test.BackendTest(t, func() backend.Backend {
 		db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@/?parseTime=true&interpolateParams=true", testUser, testPassword))
 		if err != nil {
 			panic(err)
 		}
 
+		dbName = "test_" + strings.Replace(uuid.NewString(), "-", "", -1)
 		if _, err := db.Exec("CREATE DATABASE " + dbName); err != nil {
 			panic(fmt.Errorf("creating database: %w", err))
 		}

backend/redis/instance.go

@@ -79,7 +79,12 @@ func (rb *redisBackend) GetWorkflowInstanceState(ctx context.Context, instance *
 }
 
 func (rb *redisBackend) CancelWorkflowInstance(ctx context.Context, instance *core.WorkflowInstance, event *history.Event) error {
-	// Recursively, find any sub-workflow instance to cancel
+	_, err := readInstance(ctx, rb.rdb, instance.InstanceID)
+	if err != nil {
+		return err
+	}
+
+	// Recursively, find any sub-workflow instances to cancel
 	toCancel := make([]*core.WorkflowInstance, 0)
 	toCancel = append(toCancel, instance)
 	for len(toCancel) > 0 {
@@ -94,7 +99,7 @@ func (rb *redisBackend) CancelWorkflowInstance(ctx context.Context, instance *co
 		// Find sub-workflows
 		subInstances, err := subWorkflowInstances(ctx, rb.rdb, instance)
 		if err != nil {
-			return err
+			return fmt.Errorf("finding sub-workflow instances for cancellation: %w", err)
 		}
 
 		toCancel = append(toCancel, subInstances...)

backend/redis/workflow.go

@@ -182,6 +182,10 @@ func (rb *redisBackend) CompleteWorkflowTask(ctx context.Context, taskID string,
 				if err := addFutureEvent(ctx, rb.rdb, targetInstance, &event); err != nil {
 					return err
 				}
+			} else if event.Type == history.EventType_SubWorkflowCancellationRequested {
+				if err := rb.CancelWorkflowInstance(ctx, targetInstance, &event); err != nil {
+					return err
+				}
 			} else {
 				// Add pending event to stream
 				lastPendingMessageID, err = addEventToStream(ctx, rb.rdb, pendingEventsKey(targetInstance.InstanceID), &event)

backend/sqlite/schema.sql

@@ -14,27 +14,29 @@ CREATE INDEX IF NOT EXISTS `idx_instances_locked_until_completed_at` ON `instanc
 CREATE INDEX IF NOT EXISTS `idx_instances_parent_instance_id` ON `instances` (`parent_instance_id`);
 
 CREATE TABLE IF NOT EXISTS `pending_events` (
-  `id` TEXT PRIMARY KEY,
+  `id` TEXT,
   `sequence_id` INTEGER NOT NULL, -- not used but keep for now for query compat
   `instance_id` TEXT NOT NULL,
   `event_type` INTEGER NOT NULL,
   `timestamp` DATETIME NOT NULL,
   `schedule_event_id` INT NOT NULL,
   `attributes` BLOB NOT NULL,
-  `visible_at` DATETIME NULL
+  `visible_at` DATETIME NULL,
+  PRIMARY KEY(`id`, `instance_id`)
 );
 
 CREATE INDEX IF NOT EXISTS `idx_pending_events_instance_id_visible_at` ON `pending_events` (`instance_id`, `visible_at`);
 
 CREATE TABLE IF NOT EXISTS `history` (
-  `id` TEXT PRIMARY KEY,
+  `id` TEXT,
   `sequence_id` INTEGER NOT NULL,
   `instance_id` TEXT NOT NULL,
   `event_type` INTEGER NOT NULL,
   `timestamp` DATETIME NOT NULL,
   `schedule_event_id` INT NOT NULL,
   `attributes` BLOB NOT NULL,
-  `visible_at` DATETIME NULL
+  `visible_at` DATETIME NULL,
+  PRIMARY KEY(`id`, `instance_id`)
 );
 
 CREATE INDEX IF NOT EXISTS `idx_history_instance_sequence_id` ON `history` (`instance_id`, `sequence_id`);

backend/sqlite/sqlite.go

@@ -133,31 +133,37 @@ func (sb *sqliteBackend) CancelWorkflowInstance(ctx context.Context, instance *w
 
 	instanceID := instance.InstanceID
 
-	// Cancel workflow instance
-	if err := insertNewEvents(ctx, tx, instanceID, []history.Event{*event}); err != nil {
-		return fmt.Errorf("inserting cancellation event: %w", err)
+	// TODO: Combine with event insertion
+	res := tx.QueryRowContext(ctx, "SELECT 1 FROM `instances` WHERE id = ? LIMIT 1", instanceID)
+	if err := res.Scan(nil); err == sql.ErrNoRows {
+		return backend.ErrInstanceNotFound
 	}
 
 	// Recursively, find any sub-workflow instance to cancel
-	for {
-		row := tx.QueryRowContext(ctx, "SELECT id FROM `instances` WHERE parent_instance_id = ? AND completed_at IS NULL LIMIT 1", instanceID)
-
-		var subWorkflowInstanceID string
-		if err := row.Scan(&subWorkflowInstanceID); err != nil {
-			if err == sql.ErrNoRows {
-				// No more sub-workflow instances to cancel
-				break
-			}
+	toCancel := []string{instance.InstanceID}
 
-			return fmt.Errorf("geting workflow instance for cancelling: %w", err)
-		}
+	for len(toCancel) > 0 {
+		toCancelID := toCancel[0]
+		toCancel = toCancel[1:]
 
-		// Cancel sub-workflow instance
-		if err := insertNewEvents(ctx, tx, subWorkflowInstanceID, []history.Event{*event}); err != nil {
+		if err := insertNewEvents(ctx, tx, toCancelID, []history.Event{*event}); err != nil {
 			return fmt.Errorf("inserting cancellation event: %w", err)
 		}
 
-		instanceID = subWorkflowInstanceID
+		rows, err := tx.QueryContext(ctx, "SELECT id FROM `instances` WHERE parent_instance_id = ? AND completed_at IS NULL", toCancelID)
+		defer rows.Close()
+		if err != nil {
+			return fmt.Errorf("finding sub-workflow instances: %w", err)
+		}
+
+		for rows.Next() {
+			var subWorkflowInstanceID string
+			if err := rows.Scan(&subWorkflowInstanceID); err != nil {
+				return fmt.Errorf("geting workflow instance for canceling: %w", err)
+			}
+
+			toCancel = append(toCancel, subWorkflowInstanceID)
+		}
 	}
 
 	return tx.Commit()