Merge pull request #344 from cschleiden/fix-diag-paging

Fix mysql pagination

Author: cschleiden

backend/mysql/diagnostics.go

@@ -25,7 +25,7 @@ func (mb *mysqlBackend) GetWorkflowInstances(ctx context.Context, afterInstanceI
 			ctx,
 			`SELECT i.instance_id, i.execution_id, i.created_at, i.completed_at
 			FROM instances i
-			INNER JOIN (SELECT instance_id, created_at FROM instances WHERE id = ? AND execution_id = ?) ii
+			INNER JOIN (SELECT instance_id, created_at FROM instances WHERE instance_id = ? AND execution_id = ?) ii
 				ON i.created_at < ii.created_at OR (i.created_at = ii.created_at AND i.instance_id < ii.instance_id)
 			ORDER BY i.created_at DESC, i.instance_id DESC
 			LIMIT ?`,

backend/redis/diagnostics.go

@@ -17,10 +17,10 @@ func (rb *redisBackend) GetWorkflowInstances(ctx context.Context, afterInstanceI
 	max := "+inf"
 
 	if afterInstanceID != "" {
-		afterID := instanceSegment(core.NewWorkflowInstance(afterInstanceID, afterExecutionID))
-		scores, err := rb.rdb.ZMScore(ctx, rb.keys.instancesByCreation(), afterID).Result()
+		afterSegmentID := instanceSegment(core.NewWorkflowInstance(afterInstanceID, afterExecutionID))
+		scores, err := rb.rdb.ZMScore(ctx, rb.keys.instancesByCreation(), afterSegmentID).Result()
 		if err != nil {
-			return nil, fmt.Errorf("getting instance score for %v: %w", afterID, err)
+			return nil, fmt.Errorf("getting instance score for %v: %w", afterSegmentID, err)
 		}
 
 		if len(scores) == 0 {

backend/redis/instance.go

@@ -58,7 +58,7 @@ func (rb *redisBackend) CreateWorkflowInstance(ctx context.Context, instance *wo
 		event.ID,
 		eventData,
 		payloadData,
-		time.Now().UTC().Unix(),
+		time.Now().UTC().UnixNano(),
 	).Result()
 
 	if err != nil {

backend/test/e2e.go

@@ -781,6 +781,7 @@ func EndToEndBackendTest(t *testing.T, setup func(options ...backend.BackendOpti
 	tests = append(tests, e2eActivityTests...)
 	tests = append(tests, e2eTimerTests...)
 	tests = append(tests, e2eStatsTests...)
+	tests = append(tests, e2eDiagTests...)
 
 	run := func(suffix string, workerOptions worker.Options) {
 		for _, tt := range tests {

backend/test/e2e_diag.go

@@ -0,0 +1,53 @@
+package test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cschleiden/go-workflows/client"
+	"github.com/cschleiden/go-workflows/diag"
+	"github.com/cschleiden/go-workflows/worker"
+	"github.com/cschleiden/go-workflows/workflow"
+	"github.com/stretchr/testify/require"
+)
+
+var e2eDiagTests = []backendTest{
+	{
+		name: "Diag_Paging",
+		f: func(t *testing.T, ctx context.Context, c *client.Client, w *worker.Worker, b TestBackend) {
+			diagBackend, ok := b.(diag.Backend)
+			if !ok {
+				t.Skip("Backend does not implement diag.Backend")
+			}
+
+			wf := func(ctx workflow.Context) (bool, error) {
+				return true, nil
+			}
+
+			register(t, ctx, w, []interface{}{wf}, nil)
+
+			for i := 0; i < 50; i++ {
+				runWorkflow(t, ctx, c, wf)
+			}
+
+			afterInstanceID := ""
+			afterExecutionID := ""
+
+			// Fetch 5 pages
+			for i := 0; i < 5; i++ {
+				refs, err := diagBackend.GetWorkflowInstances(ctx, afterInstanceID, afterExecutionID, 10)
+				require.NoError(t, err)
+				require.Len(t, refs, 10)
+
+				require.NotEqual(t, afterInstanceID, refs[len(refs)-1].Instance.InstanceID)
+
+				afterInstanceID = refs[len(refs)-1].Instance.InstanceID
+				afterExecutionID = refs[len(refs)-1].Instance.ExecutionID
+			}
+
+			refs, err := diagBackend.GetWorkflowInstances(ctx, afterInstanceID, afterExecutionID, 10)
+			require.NoError(t, err)
+			require.Len(t, refs, 0)
+		},
+	},
+}