Merge pull request #195 from cschleiden/remove-instances

Support removing workflow instances

Author: cschleiden

README.md

@@ -236,6 +236,19 @@ wf, err := c.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
 	InstanceID: uuid.NewString(),
 }, Workflow1, "input-for-workflow")
 if err != nil {
+	// ...
+}
+```
+
+### Removing workflow instances
+
+`RemoveWorkflowInstance` on a client instance will remove that workflow instance including all history data from the backend. A workflow instance needs to be in the finished state before calling this, otherwise an error will be returned.
+
+```go
+err = c.RemoveWorkflowInstance(ctx, workflowInstance)
+if err != nil {
+	// ...
+}
 ```
 
 ### Canceling workflows

backend/backend.go

@@ -16,6 +16,7 @@ import (
 
 var ErrInstanceNotFound = errors.New("workflow instance not found")
 var ErrInstanceAlreadyExists = errors.New("workflow instance already exists")
+var ErrInstanceNotFinished = errors.New("workflow instance is not finished")
 
 const TracerName = "go-workflow"
 
@@ -27,6 +28,9 @@ type Backend interface {
 	// CancelWorkflowInstance cancels a running workflow instance
 	CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, cancelEvent *history.Event) error
 
+	// RemoveWorkflowInstance removes a workflow instance
+	RemoveWorkflowInstance(ctx context.Context, instance *workflow.Instance) error
+
 	// GetWorkflowInstanceState returns the state of the given workflow instance
 	GetWorkflowInstanceState(ctx context.Context, instance *workflow.Instance) (core.WorkflowInstanceState, error)
 

backend/mock_Backend.go

@@ -62,7 +62,22 @@ type mysqlBackend struct {
 	options    backend.Options
 }
 
-// CreateWorkflowInstance creates a new workflow instance
+func (b *mysqlBackend) Logger() log.Logger {
+	return b.options.Logger
+}
+
+func (b *mysqlBackend) Tracer() trace.Tracer {
+	return b.options.TracerProvider.Tracer(backend.TracerName)
+}
+
+func (b *mysqlBackend) Metrics() metrics.Client {
+	return b.options.Metrics.WithTags(metrics.Tags{metrickeys.Backend: "mysql"})
+}
+
+func (b *mysqlBackend) Converter() converter.Converter {
+	return b.options.Converter
+}
+
 func (b *mysqlBackend) CreateWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {
 	tx, err := b.db.BeginTx(ctx, &sql.TxOptions{
 		Isolation: sql.LevelReadCommitted,
@@ -89,20 +104,37 @@ func (b *mysqlBackend) CreateWorkflowInstance(ctx context.Context, instance *wor
 	return nil
 }
 
-func (b *mysqlBackend) Logger() log.Logger {
-	return b.options.Logger
-}
+func (b *mysqlBackend) RemoveWorkflowInstance(ctx context.Context, instance *core.WorkflowInstance) error {
+	tx, err := b.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
 
-func (b *mysqlBackend) Tracer() trace.Tracer {
-	return b.options.TracerProvider.Tracer(backend.TracerName)
-}
+	instanceID := instance.InstanceID
 
-func (b *mysqlBackend) Metrics() metrics.Client {
-	return b.options.Metrics.WithTags(metrics.Tags{metrickeys.Backend: "mysql"})
-}
+	row := tx.QueryRowContext(ctx, "SELECT completed_at FROM `instances` WHERE instance_id = ? LIMIT 1", instanceID)
+	var completedAt sql.NullTime
+	if err := row.Scan(&completedAt); err != nil {
+		if err == sql.ErrNoRows {
+			return backend.ErrInstanceNotFound
+		}
+	}
 
-func (b *mysqlBackend) Converter() converter.Converter {
-	return b.options.Converter
+	if !completedAt.Valid {
+		return backend.ErrInstanceNotFinished
+	}
+
+	// Delete from instances and history tables
+	if _, err := tx.ExecContext(ctx, "DELETE FROM `instances` WHERE instance_id = ?", instanceID); err != nil {
+		return err
+	}
+
+	if _, err := tx.ExecContext(ctx, "DELETE FROM `history` WHERE instance_id = ?", instanceID); err != nil {
+		return err
+	}
+
+	return tx.Commit()
 }
 
 func (b *mysqlBackend) CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {

backend/mysql/mysql.go

@@ -0,0 +1,34 @@
+package redis
+
+import (
+	"context"
+	"fmt"
+
+	redis "github.com/redis/go-redis/v9"
+)
+
+// KEYS[1] - instance key
+// KEYS[2] - pending events key
+// KEYS[3] - history key
+// KEYS[4] - instances-by-creation key
+// ARGV[1] - instance ID
+var deleteCmd = redis.NewScript(
+	`redis.call("DEL", KEYS[1], KEYS[2], KEYS[3])
+	return redis.call("ZREM", KEYS[4], ARGV[1])`)
+
+// deleteInstance deletes an instance from Redis. It does not attempt to remove any future events or pending
+// workflow tasks. It's assumed that the instance is in the finished state.
+//
+// Note: might want to revisit this in the future if we want to support removing hung instances.
+func deleteInstance(ctx context.Context, rdb redis.UniversalClient, instanceID string) error {
+	if err := deleteCmd.Run(ctx, rdb, []string{
+		instanceKey(instanceID),
+		pendingEventsKey(instanceID),
+		historyKey(instanceID),
+		instancesByCreation(),
+	}, instanceID).Err(); err != nil {
+		return fmt.Errorf("failed to delete instance: %w", err)
+	}
+
+	return nil
+}

backend/redis/delete.go

@@ -109,6 +109,20 @@ func (rb *redisBackend) CancelWorkflowInstance(ctx context.Context, instance *co
 	return nil
 }
 
+func (rb *redisBackend) RemoveWorkflowInstance(ctx context.Context, instance *core.WorkflowInstance) error {
+	i, err := readInstance(ctx, rb.rdb, instance.InstanceID)
+	if err != nil {
+		return err
+	}
+
+	// Check state
+	if i.State != core.WorkflowInstanceStateFinished {
+		return backend.ErrInstanceNotFinished
+	}
+
+	return deleteInstance(ctx, rb.rdb, instance.InstanceID)
+}
+
 type instanceState struct {
 	Instance *core.WorkflowInstance     `json:"instance,omitempty"`
 	State    core.WorkflowInstanceState `json:"state,omitempty"`

backend/redis/instance.go

@@ -8,6 +8,8 @@ func instanceKey(instanceID string) string {
 	return fmt.Sprintf("instance:%v", instanceID)
 }
 
+// instancesByCreation returns the key for the ZSET that contains all instances sorted by creation date. The score is the
+// creation time. Used for listing all workflow instances in the diagnostics UI.
 func instancesByCreation() string {
 	return "instances-by-creation"
 }

backend/redis/keys.go

@@ -69,7 +69,7 @@ func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (
 		activityQueue: activityQueue,
 	}
 
-	// Preload scripts here. Usually redis-go attempts to execute them first, and the if redis doesn't know
+	// Preload scripts here. Usually redis-go attempts to execute them first, and if redis doesn't know
 	// them, loads them. This doesn't work when using (transactional) pipelines, so eagerly load them on startup.
 	ctx := context.Background()
 	cmds := map[string]*redis.StringCmd{
@@ -79,6 +79,7 @@ func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (
 		"removeFutureEventCmd":   removeFutureEventCmd.Load(ctx, rb.rdb),
 		"removePendingEventsCmd": removePendingEventsCmd.Load(ctx, rb.rdb),
 		"requeueInstanceCmd":     requeueInstanceCmd.Load(ctx, rb.rdb),
+		"deleteInstanceCmd":      deleteCmd.Load(ctx, rb.rdb),
 	}
 	for name, cmd := range cmds {
 		// fmt.Println(name, cmd.Val())

backend/redis/redis.go

@@ -64,6 +64,8 @@ type sqliteBackend struct {
 	options    backend.Options
 }
 
+var _ backend.Backend = (*sqliteBackend)(nil)
+
 func (sb *sqliteBackend) Logger() log.Logger {
 	return sb.options.Logger
 }
@@ -145,6 +147,39 @@ func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, met
 	return nil
 }
 
+func (sb *sqliteBackend) RemoveWorkflowInstance(ctx context.Context, instance *core.WorkflowInstance) error {
+	tx, err := sb.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	instanceID := instance.InstanceID
+
+	row := tx.QueryRowContext(ctx, "SELECT completed_at FROM `instances` WHERE id = ? LIMIT 1", instanceID)
+	var completedAt sql.NullTime
+	if err := row.Scan(&completedAt); err != nil {
+		if err == sql.ErrNoRows {
+			return backend.ErrInstanceNotFound
+		}
+	}
+
+	if !completedAt.Valid {
+		return backend.ErrInstanceNotFinished
+	}
+
+	// Delete from instances and history tables
+	if _, err := tx.ExecContext(ctx, "DELETE FROM `instances` WHERE id = ?", instanceID); err != nil {
+		return err
+	}
+
+	if _, err := tx.ExecContext(ctx, "DELETE FROM `history` WHERE instance_id = ?", instanceID); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
 func (sb *sqliteBackend) CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {
 	tx, err := sb.db.BeginTx(ctx, nil)
 	if err != nil {

backend/sqlite/sqlite.go

@@ -81,6 +81,33 @@ func BackendTest(t *testing.T, setup func() TestBackend, teardown func(b TestBac
 				require.Equal(t, *metadata, *task.Metadata)
 			},
 		},
+		{
+			name: "RemoveWorkflowInstance_ErrorWhenInstanceInProgress",
+			f: func(t *testing.T, ctx context.Context, b backend.Backend) {
+				instanceID := uuid.NewString()
+				wfi := core.NewWorkflowInstance(instanceID)
+
+				err := b.CreateWorkflowInstance(
+					ctx, wfi, history.NewHistoryEvent(1, time.Now(), history.EventType_WorkflowExecutionStarted, &history.ExecutionStartedAttributes{}),
+				)
+				require.NoError(t, err)
+
+				err = b.RemoveWorkflowInstance(ctx, wfi)
+				require.Error(t, err)
+				require.Equal(t, backend.ErrInstanceNotFinished, err)
+			},
+		},
+		{
+			name: "RemoveWorkflowInstance_ErrorWhenInstanceDoesNotExist",
+			f: func(t *testing.T, ctx context.Context, b backend.Backend) {
+				instanceID := uuid.NewString()
+				wfi := core.NewWorkflowInstance(instanceID)
+
+				err := b.RemoveWorkflowInstance(ctx, wfi)
+				require.Error(t, err)
+				require.Equal(t, backend.ErrInstanceNotFound, err)
+			},
+		},
 		{
 			name: "GetWorkflowTask_ReturnsNilWhenTimeout",
 			f: func(t *testing.T, ctx context.Context, b backend.Backend) {