qeueue decode + test

Author: maxdml

dbos/queue.go

@@ -1,10 +1,7 @@
 package dbos
 
 import (
-	"bytes"
 	"context"
-	"encoding/base64"
-	"encoding/gob"
 	"log/slog"
 	"math"
 	"math/rand"
@@ -227,23 +224,13 @@ func (qr *queueRunner) run(ctx *dbosContext) {
 					continue
 				}
 
-				// Deserialize input
-				var input any
-				if len(workflow.input) > 0 {
-					inputBytes, err := base64.StdEncoding.DecodeString(workflow.input)
-					if err != nil {
-						qr.logger.Error("failed to decode input for workflow", "workflow_id", workflow.id, "error", err)
-						continue
-					}
-					buf := bytes.NewBuffer(inputBytes)
-					dec := gob.NewDecoder(buf)
-					if err := dec.Decode(&input); err != nil {
-						qr.logger.Error("failed to decode input for workflow", "workflow_id", workflow.id, "error", err)
-						continue
-					}
+				input, err := ctx.serializer.Decode(&workflow.input)
+				if err != nil {
+					qr.logger.Error("Failed to decode workflow input", "workflow_id", workflow.id, "error", err)
+					continue
 				}
 
-				_, err := registeredWorkflow.wrappedFunction(ctx, input, WithWorkflowID(workflow.id))
+				_, err = registeredWorkflow.wrappedFunction(ctx, input, WithWorkflowID(workflow.id))
 				if err != nil {
 					qr.logger.Error("Error running queued workflow", "error", err)
 				}

dbos/serialization_test.go

@@ -534,6 +534,9 @@ func TestSerializer(t *testing.T) {
 		t.Run(serializerName, func(t *testing.T) {
 			executor := setupDBOS(t, true, true, serializerFactory())
 
+			// Create a test queue for queued workflow tests
+			testQueue := NewWorkflowQueue(executor, "serializer-test-queue")
+
 			// Register workflows
 			RegisterWorkflow(executor, serializerWorkflow)
 			RegisterWorkflow(executor, serializerNilValueWorkflow)
@@ -747,6 +750,56 @@ func TestSerializer(t *testing.T) {
 				testWorkflowRecovery(t, executor, serializerAnyRecoveryWorkflow, serializerAnyRecoveryStartEvent, serializerAnyRecoveryEvent, any(input), "serializer-any-recovery-wf")
 			})
 
+			// Test queued workflow with TestWorkflowData type
+			t.Run("QueuedWorkflow", func(t *testing.T) {
+				input := TestWorkflowData{
+					ID:       "queued-test-id",
+					Message:  "queued test message",
+					Value:    456,
+					Active:   false,
+					Data:     TestData{Message: "queued nested", Value: 789, Active: true},
+					Metadata: map[string]string{"type": "queued"},
+				}
+
+				// Start workflow with queue option
+				handle, err := RunWorkflow(executor, serializerWorkflow, input, WithWorkflowID("serializer-queued-wf"), WithQueue(testQueue.Name))
+				require.NoError(t, err, "failed to start queued workflow")
+
+				// Get result from the handle
+				result, err := handle.GetResult()
+				require.NoError(t, err, "queued workflow should complete successfully")
+				assert.Equal(t, input, result, "queued workflow result should match input")
+			})
+
+			// Test queued workflow with any type
+			t.Run("QueuedWorkflowAny", func(t *testing.T) {
+				if serializerName == "Gob" {
+					t.Skip("Skipping test for Gob serializer due to Gob limitations with interface types")
+				}
+
+				input := TestWorkflowData{
+					ID:       "queued-any-test-id",
+					Message:  "queued any test message",
+					Value:    321,
+					Active:   true,
+					Data:     TestData{Message: "queued any nested", Value: 654, Active: false},
+					Metadata: map[string]string{"type": "queued-any"},
+				}
+
+				// Start workflow with queue option
+				handle, err := RunWorkflow(executor, serializerAnyValueWorkflow, any(input), WithWorkflowID("serializer-queued-any-wf"), WithQueue(testQueue.Name))
+				require.NoError(t, err, "failed to start queued workflow")
+
+				// Get result from the handle
+				result, err := handle.GetResult()
+				require.NoError(t, err, "queued workflow should complete successfully")
+
+				// Convert the result from any type
+				typedResult, err := convertJSONToType[TestWorkflowData](result)
+				require.NoError(t, err, "Failed to convert result")
+				assert.Equal(t, input, typedResult, "queued workflow result should match input")
+			})
+
 		})
 	}
 }

dbos/workflow.go

@@ -990,7 +990,6 @@ func (c *dbosContext) RunWorkflow(_ DBOSContext, fn WorkflowFunc, input any, opt
 
 		// Handle DBOS ID conflict errors by waiting workflow result
 		needsConversion := false
-		fmt.Println(err)
 		if errors.Is(err, &DBOSError{Code: ConflictingIDError}) {
 			fmt.Println("Detected workflow ID conflict error, awaiting existing workflow result")
 			c.logger.Warn("Workflow ID conflict detected. Waiting for existing workflow to complete", "workflow_id", workflowID)