Add a sqlite backend option for auto vacuum.

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

@@ -246,41 +246,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()
@@ -297,23 +298,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

@@ -248,15 +248,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()
 
@@ -267,7 +267,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
@@ -279,7 +279,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 {
@@ -290,11 +290,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))]
@@ -311,25 +312,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: 500,
 }
 
 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/options.go

@@ -9,6 +9,16 @@ type options struct {
 
 	// ApplyMigrations automatically applies database migrations on startup.
 	ApplyMigrations bool
+
+	// AutoVacuum runs the `PRAGMA auto_vacuum=full` when creating the connection to enable the sqlite auto-vacuum feature.
+	//
+	// The `VACUUM` statement is always run after enabling auto-vacuum to ensure auto-vacuum is correctly enabled and to
+	// reorganize the database file and reclaim disk space.
+	//
+	// See
+	// - https://sqlite.org/pragma.html#pragma_auto_vacuum
+	// - https://sqlite.org/lang_vacuum.html.
+	AutoVacuum bool
 }
 
 type option func(*options)
@@ -28,3 +38,10 @@ func WithBackendOptions(opts ...backend.BackendOption) option {
 		}
 	}
 }
+
+// WithAutoVacuum sets sqlite auto-vacuum to full. See options.AutoVacuum for details.
+func WithAutoVacuum() option {
+	return func(o *options) {
+		o.AutoVacuum = true
+	}
+}