go-workflows

In go-workflows, changing the logic of a workflow function (e.g., adding or removing an activity) without versioning will cause non-recoverable errors during history replay. This happens because the workflow engine attempts to match the new code logic against the existing event history, and a mismatch (like an unexpected ActivitySchedule event) triggers an error.

While other platforms like Cadence or Temporal use a workflow.Version(ctx) check to handle logic branches, go-workflows does not currently support built-in workflow versioning.

To avoid replay errors when updating workflow logic, use one of the following strategies:

1. Side-by-side deployments: Run the new version of your workflow alongside the old version.
2. Queue-based routing: Use different Queues to route specific workflow versions to specific workers.

// Example of what causes a replay error:
// If you change Workflow1 from version 1 to version 2 by adding Activity3,
// a running workflow that already completed Activity1 and is waiting for Activity2
// will fail because Activity3 is not in its recorded history.

func Workflow1(ctx workflow.Context) {
	r1, _ := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1, 35, 12).Get(ctx)
	log.Println("A1 result:", r1)

	// Adding this line breaks existing workflows in replay:
	r3, _ := workflow.ExecuteActivity(ctx, workflow.DefaultActivityOptions, Activity3).Get(ctx)
	log.Println("A3 result:", r3)

	r2, _ := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity2).Get(ctx)
	log.Println("A2 result:", r2)
}

In go-workflows, the system is divided into three main components: Workflows, Activities, and Workers.

• Workflows: These define the orchestration logic. They must be deterministic; you cannot use non-deterministic Go features like select statements or iterating over maps, as the workflow state is reconstructed by replaying the execution. Inputs and outputs must be serializable.
• Activities: These are the units of work that perform side effects (e.g., API calls, database writes). Unlike workflows, activities do not need to be deterministic. They are executed once, and their results are persisted by the backend.
• Workers: These are the execution engines. A worker must have both the workflows and the activities it is intended to run registered with it. It connects to a Backend to receive tasks and report progress.

// Workflow: Orchestration (Must be deterministic)
func Workflow1(ctx workflow.Context, input string) error {
	// Use workflow.ExecuteActivity to call activities
	r1, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1, 35, 12).Get(ctx)
	if err != nil {
		panic("error getting activity 1 result")
	}
	return nil
}

// Activity: Side-effects (Can be non-deterministic)
func Activity1(ctx context.Context, a, b int) (int, error) {
	return a + b, nil
}

The Redis backend uses several different Redis data structures to manage workflow state, history, timers, and task queues. Understanding these structures is helpful for debugging or monitoring the system state:

• Workflow Instance State: Stored as JSON blobs under keys following the pattern instances-{instanceID}. This includes metadata like started_at and completed_at.
• Workflow History: Events for a specific instance are stored in Redis Streams under the key events-{instanceID}. The system maintains a cursor in the instance state to track the last executed event; any events following this cursor in the stream are treated as pending.
• Timer Events: Managed using a Redis Sorted Set (ZSET). The system checks this set during task polling to identify timer events that have reached their scheduled time, which are then added to the pending events list.
• Task Queues: Implemented using Redis STREAMs for both activities and workflow instances. To ensure task uniqueness in certain queues, an additional Redis SET is maintained.

You can create a SQLite backend using NewSqliteBackend for a fresh instance or NewSqliteBackendWithDB to use an existing *sql.DB connection.

Using an existing connection

When using NewSqliteBackendWithDB:

• The backend will not close the database connection when Close() is called.
• Migrations are disabled by default. Use WithApplyMigrations(true) to enable them.
• You are responsible for configuring the connection (e.g., WAL mode, busy timeout, max open connections).

SQLite Options

• WithApplyMigrations(applyMigrations bool): Set whether migrations should be applied on startup. Defaults to true for NewSqliteBackend and false for NewSqliteBackendWithDB.
• WithBackendOptions(opts ...backend.BackendOption): Apply generic backend options.

db, _ := sql.Open("sqlite", "file:mydb.sqlite?_txlock=immediate")
db.Exec("PRAGMA journal_mode=WAL;")
db.Exec("PRAGMA busy_timeout = 5000;")
db.SetMaxOpenConns(1)

backend := sqlite.NewSqliteBackendWithDB(db, sqlite.WithApplyMigrations(true))