Auto expire finished workflow instances

Author: cschleiden

README.md

@@ -145,7 +145,8 @@ b := mysql.NewMysqlBackend("localhost", 3306, "root", "SqlPassw0rd", "simple")
 #### Redis
 
 ```go
-b, err := redis.NewRedisBackend("localhost:6379", "user", "RedisPassw0rd", 0)
+redisClient := ...
+b, err := redis.NewRedisBackend(redisClient)
 if err != nil {
 	panic(err)
 }
@@ -251,6 +252,15 @@ if err != nil {
 }
 ```
 
+#### Automatically expiring finished workflow instances
+
+For now this is only supported for the Redis backend. When an `AutoExpiration` is passed to the backend, finished workflow instances will be automatically removed after the specified duration. This works by setting a TTL on the Redis keys for finished workflow instances. If `AutoExpiration` is set to `0` (the default), no TTL will be set.
+
+```go
+b, err := redis.NewRedisBackend(redisClient, redis.WithAutoExpiration(time.Hour * 48))
+// ...
+```
+
 ### Canceling workflows
 
 Create a `Client` instance then then call `CancelWorkflow` to cancel a workflow. When a workflow is canceled, it's workflow context is canceled. Any subsequent calls to schedule activities or sub-workflows will immediately return an error, skipping their execution. Any activities already running when a workflow is canceled will still run to completion and their result will be available.

backend/redis/expire.go

@@ -0,0 +1,63 @@
+package redis
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	redis "github.com/redis/go-redis/v9"
+)
+
+// We can't have events for redis..we do not want to do it in-process..
+
+// Set the given expiration time on all keys passed in
+// KEYS[1] - instances-by-creation key
+// KEYS[2] - instances-expiring key
+// KEYS[3] - instance key
+// KEYS[4] - pending events key
+// KEYS[5] - history key
+// ARGV[1] - current timestamp
+// ARGV[2] - expiration time in seconds
+// ARGV[3] - expiration timestamp in unix milliseconds
+// ARGV[4] - instance ID
+var expireCmd = redis.NewScript(
+	`-- Find instances which have already expired and remove from the index set
+	local expiredInstances = redis.call("ZRANGE", KEYS[2], "-inf", ARGV[1], "BYSCORE")
+	for i = 1, #expiredInstances do
+		local instanceID = expiredInstances[i]
+		redis.call("ZREM", KEYS[1], instanceID) -- index set
+		redis.call("ZREM", KEYS[2], instanceID) -- expiration set
+	end
+
+	-- Add expiration time for future cleanup
+	redis.call("ZADD", KEYS[2], ARGV[3], ARGV[4])
+
+	-- Set expiration on all keys
+	for i = 3, #KEYS do
+		redis.call("EXPIRE", KEYS[i], ARGV[2])
+	end
+
+	return 0
+	`,
+)
+
+func setWorkflowInstanceExpiration(ctx context.Context, rdb redis.UniversalClient, instanceID string, expiration time.Duration) error {
+	now := time.Now().UnixMilli()
+	nowStr := strconv.FormatInt(now, 10)
+
+	exp := time.Now().Add(expiration).UnixMilli()
+	expStr := strconv.FormatInt(exp, 10)
+
+	return expireCmd.Run(ctx, rdb, []string{
+		instancesByCreation(),
+		instancesExpiring(),
+		instanceKey(instanceID),
+		pendingEventsKey(instanceID),
+		historyKey(instanceID),
+	},
+		nowStr,
+		expiration.Seconds(),
+		expStr,
+		instanceID,
+	).Err()
+}

backend/redis/expire_test.go

@@ -0,0 +1,53 @@
+package redis
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/client"
+	"github.com/cschleiden/go-workflows/worker"
+	"github.com/cschleiden/go-workflows/workflow"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_AutoExpiration(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+
+	autoExpirationTime := time.Second * 1
+
+	redisClient := getClient()
+	setup := getCreateBackend(redisClient, WithAutoExpiration(autoExpirationTime))
+	b := setup()
+
+	c := client.New(b)
+	w := worker.New(b, &worker.DefaultWorkerOptions)
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	require.NoError(t, w.Start(ctx))
+
+	wf := func(ctx workflow.Context) error {
+		return nil
+	}
+
+	wfi, err := c.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
+		InstanceID: uuid.NewString(),
+	}, wf)
+	require.NoError(t, err)
+
+	require.NoError(t, c.WaitForWorkflowInstance(ctx, wfi, time.Second*10))
+
+	// Wait for redis to expire the keys
+	time.Sleep(autoExpirationTime)
+
+	_, err = b.GetWorkflowInstanceState(ctx, wfi)
+	require.ErrorIs(t, err, backend.ErrInstanceNotFound)
+
+	cancel()
+	require.NoError(t, w.WaitForCompletion())
+}

backend/redis/keys.go

@@ -14,6 +14,10 @@ func instancesByCreation() string {
 	return "instances-by-creation"
 }
 
+func instancesExpiring() string {
+	return "instances-expiring"
+}
+
 func pendingEventsKey(instanceID string) string {
 	return fmt.Sprintf("pending-events:%v", instanceID)
 }

backend/redis/redis.go

@@ -58,6 +58,7 @@ func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (
 		"removePendingEventsCmd": removePendingEventsCmd.Load(ctx, rb.rdb),
 		"requeueInstanceCmd":     requeueInstanceCmd.Load(ctx, rb.rdb),
 		"deleteInstanceCmd":      deleteCmd.Load(ctx, rb.rdb),
+		"expireInstanceCmd":      expireCmd.Load(ctx, rb.rdb),
 	}
 	for name, cmd := range cmds {
 		// fmt.Println(name, cmd.Val())

backend/redis/redis_test.go

@@ -8,10 +8,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cschleiden/go-workflows/backend"
 	"github.com/cschleiden/go-workflows/backend/test"
 	"github.com/cschleiden/go-workflows/internal/history"
-	"github.com/cschleiden/go-workflows/log"
 	"github.com/redis/go-redis/v9"
 )
 
@@ -27,7 +25,7 @@ func Test_RedisBackend(t *testing.T) {
 	}
 
 	client := getClient()
-	setup := getCreateBackend(client, false)
+	setup := getCreateBackend(client)
 
 	test.BackendTest(t, setup, nil)
 }
@@ -38,7 +36,7 @@ func Test_EndToEndRedisBackend(t *testing.T) {
 	}
 
 	client := getClient()
-	setup := getCreateBackend(client, false)
+	setup := getCreateBackend(client)
 
 	test.EndToEndBackendTest(t, setup, nil)
 }
@@ -54,7 +52,7 @@ func getClient() redis.UniversalClient {
 	return client
 }
 
-func getCreateBackend(client redis.UniversalClient, ignoreLog bool) func() test.TestBackend {
+func getCreateBackend(client redis.UniversalClient, additionalOptions ...RedisBackendOption) func() test.TestBackend {
 	return func() test.TestBackend {
 		// Flush database
 		if err := client.FlushDB(context.Background()).Err(); err != nil {
@@ -74,9 +72,7 @@ func getCreateBackend(client redis.UniversalClient, ignoreLog bool) func() test.
 			WithBlockTimeout(time.Millisecond * 10),
 		}
 
-		if ignoreLog {
-			options = append(options, WithBackendOptions(backend.WithLogger(&nullLogger{})))
-		}
+		options = append(options, additionalOptions...)
 
 		b, err := NewRedisBackend(client, options...)
 		if err != nil {
@@ -87,33 +83,6 @@ func getCreateBackend(client redis.UniversalClient, ignoreLog bool) func() test.
 	}
 }
 
-type nullLogger struct {
-	defaultFields []interface{}
-}
-
-// Debug implements log.Logger
-func (*nullLogger) Debug(msg string, fields ...interface{}) {
-}
-
-// Error implements log.Logger
-func (*nullLogger) Error(msg string, fields ...interface{}) {
-}
-
-// Panic implements log.Logger
-func (*nullLogger) Panic(msg string, fields ...interface{}) {
-}
-
-// Warn implements log.Logger
-func (*nullLogger) Warn(msg string, fields ...interface{}) {
-}
-
-// With implements log.Logger
-func (nl *nullLogger) With(fields ...interface{}) log.Logger {
-	return nl
-}
-
-var _ log.Logger = (*nullLogger)(nil)
-
 var _ test.TestBackend = (*redisBackend)(nil)
 
 // GetFutureEvents

backend/redis/workflow.go

@@ -203,7 +203,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 			}
 		}
 
-		// Try to queue workflow task
+		// Try to enqueue workflow task
 		if targetInstanceID != instance.InstanceID {
 			if err := rb.workflowQueue.Enqueue(ctx, p, targetInstanceID, nil); err != nil {
 				return fmt.Errorf("enqueuing workflow task: %w", err)
@@ -273,12 +273,19 @@ func (rb *redisBackend) CompleteWorkflowTask(
 	}
 
 	if state == core.WorkflowInstanceStateFinished {
+		// Trace workflow completion
 		ctx = tracing.UnmarshalSpan(ctx, instanceState.Metadata)
 		_, span := rb.Tracer().Start(ctx, "WorkflowComplete",
 			trace.WithAttributes(
 				attribute.String("workflow_instance_id", instanceState.Instance.InstanceID),
 			))
 		span.End()
+
+		if rb.options.AutoExpiration > 0 {
+			if err := setWorkflowInstanceExpiration(ctx, rb.rdb, instance.InstanceID, rb.options.AutoExpiration); err != nil {
+				return fmt.Errorf("setting workflow instance expiration: %w", err)
+			}
+		}
 	}
 
 	return nil