Implement valkey backend

.github/workflows/go.yml

@@ -86,6 +86,46 @@ jobs:
             ${{ github.workspace }}/report.xml
         if: always()
 
+  test_valkey:
+    runs-on: ubuntu-latest
+    needs: build
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: 1.24
+          check-latest: true
+          cache: true
+
+      - name: Start Valkey (Docker)
+        run: |
+          docker run -d --name valkey -p 6379:6379 valkey/valkey:latest valkey-server --requirepass ValkeyPassw0rd
+
+      - name: Wait for Valkey readiness
+        run: |
+          for i in {1..60}; do
+            if docker exec valkey valkey-cli -a ValkeyPassw0rd PING | grep -q PONG; then
+              echo "Valkey is ready"; exit 0; fi;
+            sleep 1;
+          done
+          echo "Valkey did not become ready in time";
+          docker logs valkey || true
+          exit 1
+
+      - name: Tests (valkey backend, integration)
+        run: |
+          go test -tags=valkey_integration -timeout 240s -race -count 1 -v github.com/cschleiden/go-workflows/backend/valkey 2>&1 | go tool go-junit-report -set-exit-code -iocopy -out "${{ github.workspace }}/report.xml"
+
+      - name: Test Summary
+        uses: test-summary/action@v2
+        with:
+          paths: |
+            ${{ github.workspace }}/report.xml
+        if: always()
+
   test_sqlite:
     runs-on: ubuntu-latest
     needs: build

backend/valkey/README.md

@@ -0,0 +1,44 @@
+# Valkey backend
+
+The Valkey backend implements the go-workflows `Backend` interface using [Valkey](https://valkey.io/), an open-source Redis-compatible data store. It mirrors the Redis backend in structure and key layout.
+
+## Instances and their state
+
+Instances and their state (started_at, completed_at, etc.) are stored as JSON blobs under the `instances-{instanceID}` keys.
+
+## History and pending events
+
+Events are stored in streams per workflow instance under the `events-{instanceID}` key. A cursor in the instance state tracks the last executed event. Every event after the cursor in the stream is a pending event and will be returned to the worker in the next workflow task.
+
+## Timer events
+
+Timer events are stored in a sorted set (`ZSET`). Whenever a worker checks for a new workflow instance task, the sorted set is checked to see if any pending timer events are ready. Ready events are added to the pending events before being returned for pending workflow tasks.
+
+## Task queues
+
+Queues are needed for activities and workflow instances. In both cases tasks are enqueued, workers poll for work, and every task must eventually be processed — including recovery when a worker crashes after dequeuing but before completing.
+
+Task queues are implemented using Valkey STREAMs. For queues where only a single instance of a task should be in the queue at a time, an additional `SET` is maintained alongside the stream.
+
+<details>
+  <summary>Alternatives considered</summary>
+
+  **Option 1 - `ZSET`**:
+
+  Store queue item keys in a `ZSET` scored by the unlock timestamp. Workers query `ZRANGE` for tasks with scores in `-inf, now)` and update the score to `now + lock_timeout` on pickup. No blocking commands available.
+
+  **Option 2 - `LISTS`**:
+
+  Two `LIST`s (pending and processing) with a `ZSET` for lock expiry. Atomic task pickup via `BLMOVE`. Requires periodic scans of the processing list to recover abandoned tasks.
+
+</details>
+
+## Lua scripts
+
+All multi-key atomic operations are implemented as Lua scripts embedded in `scripts/`. The scripts use `server.call` / `server.pcall` (the Valkey-native namespace) rather than the legacy `redis.*` aliases.
+
+**Note**: The Valkey backend does not support Valkey Cluster because the Lua scripts access keys that span multiple hash slots. All keys must reside on a single Valkey node or use a key prefix that maps to the same slot.
+
+## Auto-expiration
+
+Finished workflow instances can be automatically expired after a configurable duration using `WithAutoExpiration`. Continued-as-new instances can use a separate expiry via `WithAutoExpirationContinueAsNew`. Expiration is implemented via a sorted set of expiry timestamps; cleanup runs as a side effect of workflow task completion.

backend/valkey/activity.go

@@ -0,0 +1,80 @@
+package valkey
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/backend/history"
+	"github.com/cschleiden/go-workflows/workflow"
+)
+
+func (vb *valkeyBackend) PrepareActivityQueues(ctx context.Context, queues []workflow.Queue) error {
+	return vb.activityQueue.Prepare(ctx, vb.client, queues)
+}
+
+func (vb *valkeyBackend) GetActivityTask(ctx context.Context, queues []workflow.Queue) (*backend.ActivityTask, error) {
+	activityTask, err := vb.activityQueue.Dequeue(ctx, vb.client, queues, vb.options.ActivityLockTimeout, vb.options.BlockTimeout)
+	if err != nil {
+		return nil, err
+	}
+
+	if activityTask == nil {
+		return nil, nil
+	}
+
+	return &backend.ActivityTask{
+		WorkflowInstance: activityTask.Data.Instance,
+		Queue:            workflow.Queue(activityTask.Data.Queue),
+		ID:               activityTask.TaskID,
+		ActivityID:       activityTask.Data.ID,
+		Event:            activityTask.Data.Event,
+	}, nil
+}
+
+func (vb *valkeyBackend) ExtendActivityTask(ctx context.Context, task *backend.ActivityTask) error {
+	if err := vb.activityQueue.Extend(ctx, vb.client, task.Queue, task.ID); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (vb *valkeyBackend) CompleteActivityTask(ctx context.Context, task *backend.ActivityTask, result *history.Event) error {
+	instance, err := readInstance(ctx, vb.client, vb.keys.instanceKey(task.WorkflowInstance))
+	if err != nil {
+		return err
+	}
+
+	eventData, payload, err := marshalEvent(result)
+	if err != nil {
+		return err
+	}
+
+	activityQueueKeys := vb.activityQueue.Keys(task.Queue)
+	workflowQueueKeys := vb.workflowQueue.Keys(workflow.Queue(instance.Queue))
+
+	err = vb.completeActivityTaskScript.Exec(ctx, vb.client, []string{
+		activityQueueKeys.SetKey,
+		activityQueueKeys.StreamKey,
+		vb.keys.pendingEventsKey(task.WorkflowInstance),
+		vb.keys.payloadKey(task.WorkflowInstance),
+		vb.workflowQueue.queueSetKey,
+		workflowQueueKeys.SetKey,
+		workflowQueueKeys.StreamKey,
+	}, []string{
+		task.ID,
+		vb.activityQueue.groupName,
+		result.ID,
+		eventData,
+		payload,
+		vb.workflowQueue.groupName,
+		instanceSegment(task.WorkflowInstance),
+	}).Error()
+
+	if err != nil {
+		return fmt.Errorf("completing activity task: %w", err)
+	}
+
+	return nil
+}

backend/valkey/delete.go

@@ -0,0 +1,31 @@
+package valkey
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/core"
+)
+
+// deleteInstance deletes an instance from Valkey. It does not attempt to remove any future events or pending
+// workflow tasks. It's assumed that the instance is in the finished state.
+//
+// Note: might want to revisit this in the future if we want to support removing hung instances.
+func (vb *valkeyBackend) deleteInstance(ctx context.Context, instance *core.WorkflowInstance) error {
+	err := vb.deleteInstanceScript.Exec(ctx, vb.client, []string{
+		vb.keys.instanceKey(instance),
+		vb.keys.pendingEventsKey(instance),
+		vb.keys.historyKey(instance),
+		vb.keys.payloadKey(instance),
+		vb.keys.activeInstanceExecutionKey(instance.InstanceID),
+		vb.keys.instancesByCreation(),
+	}, []string{
+		instanceSegment(instance),
+	}).Error()
+
+	if err != nil {
+		return fmt.Errorf("failed to delete instance: %w", err)
+	}
+
+	return nil
+}

backend/valkey/diagnostics.go

@@ -0,0 +1,100 @@
+package valkey
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/core"
+	"github.com/cschleiden/go-workflows/diag"
+	"github.com/cschleiden/go-workflows/internal/log"
+)
+
+var _ diag.Backend = (*valkeyBackend)(nil)
+
+func (vb *valkeyBackend) GetWorkflowInstances(ctx context.Context, afterInstanceID, afterExecutionID string, count int) ([]*diag.WorkflowInstanceRef, error) {
+	zrangeCmd := vb.client.B().Zrange().Key(vb.keys.instancesByCreation()).Min("+inf").Max("-inf").Byscore().Rev().Limit(0, int64(count))
+	if afterInstanceID != "" {
+		afterSegmentID := instanceSegment(core.NewWorkflowInstance(afterInstanceID, afterExecutionID))
+		scores, err := vb.client.Do(ctx, vb.client.B().Zmscore().Key(vb.keys.instancesByCreation()).Member(afterSegmentID).Build()).AsFloatSlice()
+		if err != nil {
+			return nil, fmt.Errorf("getting instance score for %v: %w", afterSegmentID, err)
+		}
+
+		if len(scores) == 0 {
+			vb.Options().Logger.Error("could not find instance %v",
+				log.NamespaceKey+".valkey.afterInstanceID", afterInstanceID,
+				log.NamespaceKey+".valkey.afterExecutionID", afterExecutionID,
+			)
+			return nil, nil
+		}
+
+		zrangeCmd = vb.client.B().Zrange().Key(vb.keys.instancesByCreation()).Min("+inf").Max(fmt.Sprintf("(%f", scores[0])).Byscore().Rev().Limit(0, int64(count))
+	}
+
+	instanceSegments, err := vb.client.Do(ctx, zrangeCmd.Build()).AsStrSlice()
+	if err != nil {
+		return nil, fmt.Errorf("getting instances: %w", err)
+	}
+
+	if len(instanceSegments) == 0 {
+		return nil, nil
+	}
+
+	instanceKeys := make([]string, 0)
+	for _, r := range instanceSegments {
+		instanceKeys = append(instanceKeys, vb.keys.instanceKeyFromSegment(r))
+	}
+
+	cmd := vb.client.B().Mget().Key(instanceKeys...)
+	instances, err := vb.client.Do(ctx, cmd.Build()).AsStrSlice()
+	if err != nil {
+		return nil, fmt.Errorf("getting instances: %w", err)
+	}
+
+	instanceRefs := make([]*diag.WorkflowInstanceRef, 0, len(instances))
+	for _, instance := range instances {
+		if instance == "" {
+			continue
+		}
+
+		var state instanceState
+		if err := json.Unmarshal([]byte(instance), &state); err != nil {
+			return nil, fmt.Errorf("unmarshaling instance state: %w", err)
+		}
+
+		instanceRefs = append(instanceRefs, &diag.WorkflowInstanceRef{
+			Instance:    state.Instance,
+			CreatedAt:   state.CreatedAt,
+			CompletedAt: state.CompletedAt,
+			State:       state.State,
+			Queue:       state.Queue,
+		})
+	}
+
+	return instanceRefs, nil
+}
+
+func (vb *valkeyBackend) GetWorkflowInstance(ctx context.Context, instance *core.WorkflowInstance) (*diag.WorkflowInstanceRef, error) {
+	instanceState, err := readInstance(ctx, vb.client, vb.keys.instanceKey(instance))
+	if err != nil {
+		return nil, err
+	}
+
+	return mapWorkflowInstance(instanceState), nil
+}
+
+func (vb *valkeyBackend) GetWorkflowTree(ctx context.Context, instance *core.WorkflowInstance) (*diag.WorkflowInstanceTree, error) {
+	itb := diag.NewInstanceTreeBuilder(vb)
+	return itb.BuildWorkflowInstanceTree(ctx, instance)
+}
+
+func mapWorkflowInstance(instance *instanceState) *diag.WorkflowInstanceRef {
+	return &diag.WorkflowInstanceRef{
+		Instance:    instance.Instance,
+		CreatedAt:   instance.CreatedAt,
+		CompletedAt: instance.CompletedAt,
+		State:       instance.State,
+		Queue:       instance.Queue,
+	}
+}