Merge branch 'main' into yaananth-patch-2

Author: yaananth

README.md

@@ -314,7 +314,7 @@ func Workflow2(ctx workflow.Context, msg string) (string, error) {
 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)
+r1, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1, 35, 12, nil, "test").Get(ctx)
 if err != nil {
 	panic("error getting activity 1 result")
 }
@@ -423,9 +423,86 @@ func SubWorkflow(ctx workflow.Context, msg string) (int, error) {
 
 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.
 
+### Error handling
+
+#### Custom errors
+
+Errors returned from activities and subworkflows need to be marshalled/unmarshalled by the library so they are wrapped in a `workflow.Error`. You can access the original type via the `err.Type` field. If a stacktrace was captured, you can access it via `err.Stack()`. Example (see also `samples/errors`):
+
+```go
+func handleError(ctx workflow.Context, logger log.Logger, err error) {
+	var werr *workflow.Error
+	if errors.As(err, &werr) {
+		switch werr.Type {
+		case "CustomError": // This was a `type CustomError struct...` returned by an activity/subworkflow
+			logger.Error("Custom error", "err", werr)
+			return
+		}
+
+		logger.Error("Generic workflow error", "err", werr, "stack", werr.Stack())
+		return
+	}
+
+	var perr *workflow.PanicError
+	if errors.As(err, &perr) {
+		// Activity/subworkflow ran into a panic
+		logger.Error("Panic", "err", perr, "stack", perr.Stack())
+		return
+	}
+
+	logger.Error("Generic error", "err", err)
+}
+```
+
+#### Panics
+
+A panic in an activity will be captured by the library and made available as a `workflow.PanicError` in the calling workflow. Example:
+
+
+```go
+r1, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1, "test").Get(ctx)
+if err != nil {
+	panic("error getting activity 1 result")
+}
+
+var perr *workflow.PanicError
+if errors.As(err, &perr) {
+	logger.Error("Panic", "err", perr, "stack", perr.Stack())
+	return
+}
+```
+
+#### Retries
+
+With the default `DefaultActivityOptions`, Activities are retried up to three times when they return an error. If you want to keep automatic retries, but want to avoid them when hitting certain error types, you can wrap an error with `workflow.NewPermanentError`:
+
+**Workflow**:
+
+```go
+r1, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1, "test").Get(ctx)
+if err != nil {
+	panic("error getting activity 1 result")
+}
+
+log.Println(r1)
+```
+
+**Activity**:
+
+```go
+func Activity1(ctx context.Context, name string) (int, error) {
+	if name == "test" {
+		// No need to retry in this case, the activity will aways fail with the given inputs
+		return 0, workflow.NewPermanentError(errors.New("test is not a valid name"))
+	}
+
+	return http.Do("POST", "https://example.com", name)
+}
+```
+
 ### `ContinueAsNew`
 
-```ContinueAsNew` allows you to restart workflow execution with different inputs. The purpose is to keep the history size small enough to avoid hitting size limits, running out of memory and impacting performance. It works by returning a special `error` from your workflow that contains the new inputs:
+`ContinueAsNew` allows you to restart workflow execution with different inputs. The purpose is to keep the history size small enough to avoid hitting size limits, running out of memory and impacting performance. It works by returning a special `error` from your workflow that contains the new inputs:
 
 ```go
 wf := func(ctx workflow.Context, run int) (int, error) {
@@ -756,6 +833,3 @@ and only if a workflow instance was created with a version of `>= 2` will `Activ
 
 This kind of check is understandable for simple changes, but it becomes hard and a source of bugs for more complicated workflows. Therefore for now versioning is not supported and the guidance is to rely on **side-by-side** deployments. See also Azure's [Durable Functions](https://docs.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-versioning) documentation for the same topic.
 
-### `ContinueAsNew`
-
-Both Temporal/Cadence and DTFx support `ContinueAsNew`. This essentially re-starts a running workflow as a new workflow with a new event history. This is needed for long running workflows where the history can become very large, negatively affecting performance. While `WorkflowInstance` supports an `InstanceID` and an `ExecutionID`, this feature is not yet implemented (and might not be).

backend/backend.go

@@ -68,6 +68,9 @@ type Backend interface {
 	// ExtendActivityTask extends the lock of an activity task
 	ExtendActivityTask(ctx context.Context, activityID string) error
 
+	// GetStats returns stats about the backend
+	GetStats(ctx context.Context) (*Stats, error)
+
 	// Logger returns the configured logger for the backend
 	Logger() log.Logger
 

backend/mock_Backend.go

@@ -0,0 +1,53 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/backend"
+)
+
+func (b *mysqlBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
+	s := &backend.Stats{}
+
+	tx, err := b.db.BeginTx(ctx, &sql.TxOptions{
+		Isolation: sql.LevelReadCommitted,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to start transaction: %w", err)
+	}
+	defer tx.Rollback()
+
+	row := tx.QueryRowContext(
+		ctx,
+		"SELECT COUNT(*) FROM instances i WHERE i.completed_at IS NULL",
+	)
+	if err := row.Err(); err != nil {
+		return nil, fmt.Errorf("failed to query active instances: %w", err)
+	}
+
+	var activeInstances int64
+	if err := row.Scan(&activeInstances); err != nil {
+		return nil, fmt.Errorf("failed to scan active instances: %w", err)
+	}
+
+	s.ActiveWorkflowInstances = activeInstances
+
+	// Get pending activities
+	row = tx.QueryRowContext(
+		ctx,
+		"SELECT COUNT(*) FROM activities")
+	if err := row.Err(); err != nil {
+		return nil, fmt.Errorf("failed to query active activities: %w", err)
+	}
+
+	var pendingActivities int64
+	if err := row.Scan(&pendingActivities); err != nil {
+		return nil, fmt.Errorf("failed to scan active activities: %w", err)
+	}
+
+	s.PendingActivities = pendingActivities
+
+	return s, nil
+}

backend/mysql/stats.go

@@ -160,6 +160,8 @@ func createInstanceP(ctx context.Context, p redis.Pipeliner, instance *core.Work
 		Score:  float64(createdAt.UnixMilli()),
 	})
 
+	p.SAdd(ctx, instancesActive(), instanceSegment(instance))
+
 	return nil
 }
 
@@ -173,6 +175,10 @@ func updateInstanceP(ctx context.Context, p redis.Pipeliner, instance *core.Work
 
 	p.Set(ctx, key, string(b), 0)
 
+	if state.State != core.WorkflowInstanceStateActive {
+		p.SRem(ctx, instancesActive(), instanceSegment(instance))
+	}
+
 	// CreatedAt does not change, so skip updating the instancesByCreation() ZSET
 
 	return nil

backend/redis/instance.go

@@ -29,6 +29,10 @@ func instancesByCreation() string {
 	return "instances-by-creation"
 }
 
+func instancesActive() string {
+	return "instances-active"
+}
+
 func instancesExpiring() string {
 	return "instances-expiring"
 }

backend/redis/keys.go

@@ -43,20 +43,11 @@ func newTaskQueue[T any](rdb redis.UniversalClient, tasktype string) (*taskQueue
 		workerName: uuid.NewString(),
 	}
 
-	// Create the consumer group
-	_, err := rdb.XGroupCreateMkStream(context.Background(), tq.streamKey, tq.groupName, "0").Result()
-	if err != nil {
-		// Ugly, check since there is no UPSERT for consumer groups. Might replace with a script
-		// using XINFO & XGROUP CREATE atomically
-		if err.Error() != "BUSYGROUP Consumer Group name already exists" {
-			return nil, fmt.Errorf("creating task queue: %w", err)
-		}
-	}
-
 	// Pre-load script
 	cmds := map[string]*redis.StringCmd{
-		"enqueueCmd":  enqueueCmd.Load(context.Background(), rdb),
-		"completeCmd": completeCmd.Load(context.Background(), rdb),
+		"createGroupCmd": createGroupCmd.Load(context.Background(), rdb),
+		"enqueueCmd":     enqueueCmd.Load(context.Background(), rdb),
+		"completeCmd":    completeCmd.Load(context.Background(), rdb),
 	}
 
 	for name, cmd := range cmds {
@@ -67,6 +58,12 @@ func newTaskQueue[T any](rdb redis.UniversalClient, tasktype string) (*taskQueue
 		}
 	}
 
+	// Create the consumer group
+	err := createGroupCmd.Run(context.Background(), rdb, []string{tq.streamKey, tq.groupName}).Err()
+	if err != nil {
+		return nil, fmt.Errorf("creating task queue: %w", err)
+	}
+
 	return tq, nil
 }
 
@@ -77,6 +74,10 @@ func (q *taskQueue[T]) Keys() KeyInfo {
 	}
 }
 
+func (q *taskQueue[T]) Size(ctx context.Context, rdb redis.UniversalClient) (int64, error) {
+	return rdb.XLen(ctx, q.streamKey).Result()
+}
+
 // KEYS[1] = set
 // KEYS[2] = stream
 // ARGV[1] = caller provided id of the task
@@ -91,6 +92,34 @@ var enqueueCmd = redis.NewScript(
 	return true
 `)
 
+var createGroupCmd = redis.NewScript(`
+    local streamKey = KEYS[1]
+    local groupName = KEYS[2]
+    local exists = false
+    local res = redis.pcall('XINFO', 'GROUPS', streamKey)
+
+    if res and type(res) == 'table' then
+        for _, groupInfo in ipairs(res) do
+            if type(groupInfo) == 'table' then
+                for i = 1, #groupInfo, 2 do
+                    if groupInfo[i] == 'name' and groupInfo[i+1] == groupName then
+                        exists = true
+                        break
+                    end
+                end
+            end
+            if exists then break end
+        end
+    end
+
+    if not exists then
+        redis.call('XGROUP', 'CREATE', streamKey, groupName, '$', 'MKSTREAM')
+    end
+
+    return true
+`)
+
+
 func (q *taskQueue[T]) Enqueue(ctx context.Context, p redis.Pipeliner, id string, data *T) error {
 	ds, err := json.Marshal(data)
 	if err != nil {

backend/redis/queue.go

@@ -0,0 +1,32 @@
+package redis
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/backend"
+)
+
+func (rb *redisBackend) GetStats(ctx context.Context) (*backend.Stats, error) {
+	var err error
+
+	s := &backend.Stats{}
+
+	// get workflow instances
+	activeInstances, err := rb.rdb.SCard(ctx, instancesActive()).Result()
+	if err != nil {
+		return nil, fmt.Errorf("getting active instances: %w", err)
+	}
+
+	s.ActiveWorkflowInstances = activeInstances
+
+	// get pending activities
+	pendingActivities, err := rb.activityQueue.Size(ctx, rb.rdb)
+	if err != nil {
+		return nil, fmt.Errorf("getting active activities: %w", err)
+	}
+
+	s.PendingActivities = pendingActivities
+
+	return s, nil
+}

backend/redis/stats.go

@@ -83,8 +83,8 @@ func (sb *sqliteBackend) Converter() converter.Converter {
 	return sb.options.Converter
 }
 
-func (b *sqliteBackend) ContextPropagators() []contextpropagation.ContextPropagator {
-	return b.options.ContextPropagators
+func (sb *sqliteBackend) ContextPropagators() []contextpropagation.ContextPropagator {
+	return sb.options.ContextPropagators
 }
 
 func (sb *sqliteBackend) CreateWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {