Update sqlite RemoveWorkflowInstances performance

backend/backend.go

@@ -37,8 +37,8 @@ type Backend interface {
 	// RemoveWorkflowInstance removes a workflow instance
 	RemoveWorkflowInstance(ctx context.Context, instance *workflow.Instance) error
 
-	// RemoveWorkflowInstances removes multiple workflow instances
-	RemoveWorkflowInstances(ctx context.Context, options ...RemovalOption) error
+	// RemoveWorkflowInstances removes multiple workflow instances and returns the number removed
+	RemoveWorkflowInstances(ctx context.Context, options ...RemovalOption) (int, error)
 
 	// GetWorkflowInstanceState returns the state of the given workflow instance
 	GetWorkflowInstanceState(ctx context.Context, instance *workflow.Instance) (core.WorkflowInstanceState, error)

backend/mysql/mysql.go

@@ -201,41 +201,42 @@ func (mb *mysqlBackend) removeWorkflowInstance(ctx context.Context, instance *co
 	return nil
 }
 
-func (mb *mysqlBackend) RemoveWorkflowInstances(ctx context.Context, options ...backend.RemovalOption) error {
+func (mb *mysqlBackend) RemoveWorkflowInstances(ctx context.Context, options ...backend.RemovalOption) (int, error) {
 	ro := backend.DefaultRemovalOptions
 	for _, opt := range options {
 		opt(&ro)
 	}
 
 	rows, err := mb.db.QueryContext(ctx, `SELECT instance_id, execution_id FROM instances WHERE completed_at < ?`, ro.FinishedBefore)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
 	instanceIDs := []string{}
 	executionIDs := []string{}
 	for rows.Next() {
 		var id, executionID string
 		if err := rows.Scan(&id, &executionID); err != nil {
-			return err
+			return 0, err
 		}
 
 		instanceIDs = append(instanceIDs, id)
 		executionIDs = append(executionIDs, executionID)
 	}
 
 	if rows.Err() != nil {
-		return rows.Err()
+		return 0, rows.Err()
 	}
 
+	removed := 0
 	batchSize := ro.BatchSize
 	for i := 0; i < len(instanceIDs); i += batchSize {
 		instanceIDs := instanceIDs[i:min(i+batchSize, len(instanceIDs))]
 		executionIDs := executionIDs[i:min(i+batchSize, len(executionIDs))]
 
 		tx, err := mb.db.BeginTx(ctx, nil)
 		if err != nil {
-			return err
+			return removed, err
 		}
 
 		defer tx.Rollback()
@@ -252,23 +253,25 @@ func (mb *mysqlBackend) RemoveWorkflowInstances(ctx context.Context, options ...
 
 		// Delete from instances, history and attributes tables
 		if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM `instances` WHERE %v", whereCondition), args...); err != nil {
-			return err
+			return removed, err
 		}
 
 		if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM `history` WHERE %v", whereCondition), args...); err != nil {
-			return err
+			return removed, err
 		}
 
 		if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM `attributes` WHERE %v", whereCondition), args...); err != nil {
-			return err
+			return removed, err
 		}
 
 		if err := tx.Commit(); err != nil {
-			return err
+			return removed, err
 		}
+
+		removed += len(instanceIDs)
 	}
 
-	return nil
+	return removed, nil
 }
 
 func (mb *mysqlBackend) CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {

backend/postgres/postgres.go

@@ -201,15 +201,15 @@ func (pb *postgresBackend) removeWorkflowInstance(ctx context.Context, instance
 	return nil
 }
 
-func (pb *postgresBackend) RemoveWorkflowInstances(ctx context.Context, options ...backend.RemovalOption) error {
+func (pb *postgresBackend) RemoveWorkflowInstances(ctx context.Context, options ...backend.RemovalOption) (int, error) {
 	ro := backend.DefaultRemovalOptions
 	for _, opt := range options {
 		opt(&ro)
 	}
 
 	rows, err := pb.db.QueryContext(ctx, "SELECT instance_id, execution_id FROM instances WHERE completed_at < $1", ro.FinishedBefore)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	defer rows.Close()
 
@@ -220,7 +220,7 @@ func (pb *postgresBackend) RemoveWorkflowInstances(ctx context.Context, options
 	for rows.Next() {
 		var id, executionID string
 		if err := rows.Scan(&id, &executionID); err != nil {
-			return err
+			return 0, err
 		}
 		pairs = append(pairs, struct {
 			instanceID  string
@@ -232,7 +232,7 @@ func (pb *postgresBackend) RemoveWorkflowInstances(ctx context.Context, options
 	}
 
 	if rows.Err() != nil {
-		return rows.Err()
+		return 0, rows.Err()
 	}
 
 	pgPairedPlaceholders := func(startIdx, pairCount int) string {
@@ -243,11 +243,12 @@ func (pb *postgresBackend) RemoveWorkflowInstances(ctx context.Context, options
 		return strings.Join(placeholders, ", ")
 	}
 
+	removed := 0
 	batchSize := ro.BatchSize
 	for i := 0; i < len(pairs); i += batchSize {
 		tx, err := pb.db.BeginTx(ctx, nil)
 		if err != nil {
-			return err
+			return removed, err
 		}
 
 		batch := pairs[i:min(i+batchSize, len(pairs))]
@@ -264,25 +265,27 @@ func (pb *postgresBackend) RemoveWorkflowInstances(ctx context.Context, options
 		// Delete from instances, history and attributes tables
 		if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM instances WHERE %v", whereCondition), args...); err != nil {
 			_ = tx.Rollback()
-			return err
+			return removed, err
 		}
 
 		if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM history WHERE %v", whereCondition), args...); err != nil {
 			_ = tx.Rollback()
-			return err
+			return removed, err
 		}
 
 		if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM attributes WHERE %v", whereCondition), args...); err != nil {
 			_ = tx.Rollback()
-			return err
+			return removed, err
 		}
 
 		if err := tx.Commit(); err != nil {
-			return err
+			return removed, err
 		}
+
+		removed += len(batch)
 	}
 
-	return nil
+	return removed, nil
 }
 
 func (pb *postgresBackend) CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {

backend/redis/instance.go

@@ -167,8 +167,8 @@ func (rb *redisBackend) RemoveWorkflowInstance(ctx context.Context, instance *co
 	return rb.deleteInstance(ctx, instance)
 }
 
-func (rb *redisBackend) RemoveWorkflowInstances(ctx context.Context, options ...backend.RemovalOption) error {
-	return backend.ErrNotSupported{
+func (rb *redisBackend) RemoveWorkflowInstances(ctx context.Context, options ...backend.RemovalOption) (int, error) {
+	return 0, backend.ErrNotSupported{
 		Message: "not supported, use auto-expiration",
 	}
 }

backend/removal.go

@@ -10,7 +10,7 @@ type RemovalOptions struct {
 }
 
 var DefaultRemovalOptions = RemovalOptions{
-	BatchSize: 100,
+	BatchSize: 50,
 }
 
 type RemovalOption func(o *RemovalOptions)

backend/sqlite/events.go

@@ -31,10 +31,6 @@ func (sb *sqliteBackend) GetFutureEvents(ctx context.Context) ([]*history.Event,
 	if err != nil {
 		return nil, fmt.Errorf("getting history: %w", err)
 	}
-	if futureEvents.Err() != nil {
-		return nil, futureEvents.Err()
-	}
-
 	defer futureEvents.Close()
 
 	f := make([]*history.Event, 0)
@@ -69,6 +65,10 @@ func (sb *sqliteBackend) GetFutureEvents(ctx context.Context) ([]*history.Event,
 		f = append(f, fe)
 	}
 
+	if futureEvents.Err() != nil {
+		return nil, futureEvents.Err()
+	}
+
 	return f, nil
 }
 
@@ -269,5 +269,5 @@ func removeFutureEvent(ctx context.Context, tx *sql.Tx, instance *core.WorkflowI
 		}
 	}
 
-	return err
+	return nil
 }

backend/sqlite/sqlite.go

@@ -254,6 +254,7 @@ func (sb *sqliteBackend) removeWorkflowInstance(ctx context.Context, instance *c
 		if err == sql.ErrNoRows {
 			return backend.ErrInstanceNotFound
 		}
+		return fmt.Errorf("scanning workflow instance state: %w", err)
 	}
 
 	if state == core.WorkflowInstanceStateActive {
@@ -275,74 +276,73 @@ func (sb *sqliteBackend) removeWorkflowInstance(ctx context.Context, instance *c
 	return nil
 }
 
-func (sb *sqliteBackend) RemoveWorkflowInstances(ctx context.Context, options ...backend.RemovalOption) error {
+func (sb *sqliteBackend) RemoveWorkflowInstances(ctx context.Context, options ...backend.RemovalOption) (int, error) {
 	ro := backend.DefaultRemovalOptions
 	for _, opt := range options {
 		opt(&ro)
 	}
 
-	rows, err := sb.db.QueryContext(ctx, `SELECT id, execution_id FROM instances WHERE completed_at < ?`, ro.FinishedBefore)
+	rows, err := sb.db.QueryContext(ctx,
+		`SELECT id, execution_id FROM instances WHERE completed_at IS NOT NULL AND completed_at < ? LIMIT ?`,
+		ro.FinishedBefore, ro.BatchSize)
 	if err != nil {
-		return err
+		return 0, err
 	}
+	defer rows.Close()
 
 	instanceIDs := []string{}
 	executionIDs := []string{}
 	for rows.Next() {
 		var id, executionID string
 		if err := rows.Scan(&id, &executionID); err != nil {
-			return err
+			return 0, err
 		}
 
 		instanceIDs = append(instanceIDs, id)
 		executionIDs = append(executionIDs, executionID)
 	}
 
 	if rows.Err() != nil {
-		return rows.Err()
+		return 0, rows.Err()
 	}
 
-	batchSize := ro.BatchSize
-	for i := 0; i < len(instanceIDs); i += batchSize {
-		instanceIDs := instanceIDs[i:min(i+batchSize, len(instanceIDs))]
-		executionIDs := executionIDs[i:min(i+batchSize, len(executionIDs))]
-
-		tx, err := sb.db.BeginTx(ctx, nil)
-		if err != nil {
-			return err
-		}
-
-		defer tx.Rollback()
+	if len(instanceIDs) == 0 {
+		return 0, nil
+	}
 
-		placeholders := strings.Repeat(",?", len(instanceIDs)-1)
-		whereCondition := fmt.Sprintf("id IN (?%v) AND execution_id IN (?%v)", placeholders, placeholders)
-		args := make([]interface{}, 0, len(instanceIDs)*2)
-		for i := range instanceIDs {
-			args = append(args, instanceIDs[i])
-		}
-		for i := range executionIDs {
-			args = append(args, executionIDs[i])
-		}
+	tx, err := sb.db.BeginTx(ctx, nil)
+	if err != nil {
+		return 0, err
+	}
 
-		// Delete from instances, history and attributes tables
-		if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM `instances` WHERE %v", whereCondition), args...); err != nil {
-			return err
-		}
+	placeholders := strings.Repeat(",?", len(instanceIDs)-1)
+	instancesWhere := fmt.Sprintf("id IN (?%v) AND execution_id IN (?%v)", placeholders, placeholders)
+	historyWhere := fmt.Sprintf("instance_id IN (?%v) AND execution_id IN (?%v)", placeholders, placeholders)
+	args := make([]interface{}, 0, len(instanceIDs)*2)
+	for j := range instanceIDs {
+		args = append(args, instanceIDs[j])
+	}
+	for j := range executionIDs {
+		args = append(args, executionIDs[j])
+	}
 
-		if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM `history` WHERE %v", whereCondition), args...); err != nil {
-			return err
-		}
+	// Delete from instances, history and attributes tables
+	if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM `instances` WHERE %v", instancesWhere), args...); err != nil {
+		_ = tx.Rollback()
+		return 0, err
+	}
 
-		if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM `attributes` WHERE %v", whereCondition), args...); err != nil {
-			return err
-		}
+	if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM `history` WHERE %v", historyWhere), args...); err != nil {
+		_ = tx.Rollback()
+		return 0, err
+	}
 
-		if err := tx.Commit(); err != nil {
-			return err
-		}
+	if _, err := tx.ExecContext(ctx, fmt.Sprintf("DELETE FROM `attributes` WHERE %v", historyWhere), args...); err != nil {
+		_ = tx.Rollback()
+		return 0, err
 	}
 
-	return nil
+	return len(instanceIDs), tx.Commit()
 }
 
 func (sb *sqliteBackend) CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {
@@ -410,6 +410,7 @@ func (sb *sqliteBackend) GetWorkflowInstanceState(ctx context.Context, instance
 		if err == sql.ErrNoRows {
 			return core.WorkflowInstanceStateActive, backend.ErrInstanceNotFound
 		}
+		return core.WorkflowInstanceStateActive, fmt.Errorf("scanning workflow instance state: %w", err)
 	}
 
 	return state, nil
@@ -425,8 +426,11 @@ func (sb *sqliteBackend) SignalWorkflow(ctx context.Context, instanceID string,
 	// TODO: Combine this with the event insertion
 	var executionID string
 	res := tx.QueryRowContext(ctx, "SELECT execution_id FROM `instances` WHERE id = ? AND state = ? LIMIT 1", instanceID, core.WorkflowInstanceStateActive)
-	if err := res.Scan(&executionID); err == sql.ErrNoRows {
-		return backend.ErrInstanceNotFound
+	if err := res.Scan(&executionID); err != nil {
+		if err == sql.ErrNoRows {
+			return backend.ErrInstanceNotFound
+		}
+		return fmt.Errorf("scanning execution ID: %w", err)
 	}
 
 	if err := insertPendingEvents(ctx, tx, core.NewWorkflowInstance(instanceID, executionID), []*history.Event{event}); err != nil {