cleanup

Author: maxdml

dbos/client.go

@@ -149,7 +149,7 @@ func (c *client) Enqueue(queueName, workflowName string, input any, opts ...Enqu
 	}
 
 	// Serialize input before storing in workflow status
-	encodedInputStr, err := serialize(dbosCtx, params.workflowInput)
+	encodedInputStr, err := serialize(dbosCtx.serializer, params.workflowInput)
 	if err != nil {
 		return nil, fmt.Errorf("failed to serialize workflow input: %w", err)
 	}
@@ -247,7 +247,6 @@ func Enqueue[P any, R any](c Client, queueName, workflowName string, input P, op
 	if c == nil {
 		return nil, errors.New("client cannot be nil")
 	}
-    
 
 	// Call the interface method with the same signature
 	handle, err := c.Enqueue(queueName, workflowName, input, opts...)

dbos/queue.go

@@ -224,13 +224,8 @@ func (qr *queueRunner) run(ctx *dbosContext) {
 					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))
+				// Pass encoded input directly - decoding will happen in workflow wrapper when we know the target type
+				_, err = registeredWorkflow.wrappedFunction(ctx, workflow.input, WithWorkflowID(workflow.id))
 				if err != nil {
 					qr.logger.Error("Error running queued workflow", "error", err)
 				}

dbos/recovery.go

@@ -14,19 +14,15 @@ func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]Workflow
 	}
 
 	for _, workflow := range pendingWorkflows {
-		// Deserialize the workflow input
-		var decodedInput any
+		// Pass encoded input directly - decoding will happen in workflow wrapper when we know the target type
+		var encodedInput *string
 		if workflow.Input != nil {
 			inputString, ok := workflow.Input.(*string)
 			if !ok {
 				ctx.logger.Warn("Skipping workflow recovery due to invalid input type", "workflow_id", workflow.ID, "name", workflow.Name, "input_type", workflow.Input)
 				continue
 			}
-			decodedInput, err = ctx.serializer.Decode(inputString)
-			if err != nil {
-				ctx.logger.Warn("Skipping workflow recovery due to input decoding failure", "workflow_id", workflow.ID, "name", workflow.Name, "error", err)
-				continue
-			}
+			encodedInput = inputString
 		}
 
 		if workflow.QueueName != "" {
@@ -63,7 +59,7 @@ func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]Workflow
 			WithWorkflowID(workflow.ID),
 		}
 		// Create a workflow context from the executor context
-		handle, err := registeredWorkflow.wrappedFunction(ctx, decodedInput, opts...)
+		handle, err := registeredWorkflow.wrappedFunction(ctx, encodedInput, opts...)
 		if err != nil {
 			return nil, err
 		}

dbos/serialization.go

@@ -23,6 +23,11 @@ func NewJSONSerializer() *JSONSerializer {
 	return &JSONSerializer{}
 }
 
+func isJSONSerializer(s Serializer) bool {
+	_, ok := s.(*JSONSerializer)
+	return ok
+}
+
 func (j *JSONSerializer) Encode(data any) (string, error) {
 	var inputBytes []byte
 	if !isNilValue(data) {
@@ -53,69 +58,53 @@ func (j *JSONSerializer) Decode(data *string) (any, error) {
 	return result, nil
 }
 
-// serialize serializes data using the serializer from the DBOSContext
-// convenience helper to also check the context & serializer
-func serialize(ctx DBOSContext, data any) (string, error) {
-	dbosCtx, ok := ctx.(*dbosContext)
-	if !ok {
-		return "", fmt.Errorf("invalid DBOSContext: expected *dbosContext")
+// serialize serializes data using the provided serializer
+func serialize[T any](serializer Serializer, data T) (string, error) {
+	if serializer == nil {
+		return "", fmt.Errorf("serializer cannot be nil")
 	}
-	if dbosCtx.serializer == nil {
-		return "", fmt.Errorf("no serializer configured in DBOSContext")
-	}
-	return dbosCtx.serializer.Encode(data)
-}
-
-func isJSONSerializer(s Serializer) bool {
-	_, ok := s.(*JSONSerializer)
-	return ok
+	return serializer.Encode(data)
 }
 
-// convertJSONToType converts a JSON-decoded value (map[string]interface{}) to type T
-// via marshal/unmarshal round-trip.
-//
-// This is needed because JSON deserialization loses type information when decoding
-// into `any` - it converts structs to map[string]interface{}, numbers to float64, etc.
-// By re-marshaling and unmarshaling into a typed target, we (mostly) restore the original structure.
-// We should be able to get rid of this when we lift encoding/decoding outside of the system database.
-func convertJSONToType[T any](value any) (T, error) {
-	if value == nil {
-		return *new(T), nil
-	}
-
-	jsonBytes, err := json.Marshal(value)
-	if err != nil {
-		return *new(T), fmt.Errorf("marshaling for type conversion: %w", err)
+// deserialize decodes an encoded string directly into a typed variable.
+// For JSON serializer, this decodes directly into the target type, preserving type information.
+// For other serializers, it decodes into any and then type-asserts.
+// (we don't want generic Serializer interface because it would require 1 serializer per type)
+func deserialize[T any](serializer Serializer, encoded *string) (T, error) {
+	if serializer == nil {
+		return *new(T), fmt.Errorf("serializer cannot be nil")
 	}
 
-	// Check if T is an interface type
 	var zero T
-	typeOfT := reflect.TypeOf(&zero).Elem()
+	if encoded == nil || *encoded == "" {
+		return zero, nil
+	}
 
-	if typeOfT.Kind() == reflect.Interface {
-		// T is interface - need to get concrete type from value
-		concreteType := reflect.TypeOf(value)
-		if concreteType.Kind() == reflect.Pointer {
-			concreteType = concreteType.Elem()
+	if isJSONSerializer(serializer) {
+		// For JSON serializer, decode directly into the target type to preserve type information
+		// We cannot just use the serializer's Decode method and recast -- the type inormation would be lost
+		dataBytes, err := base64.StdEncoding.DecodeString(*encoded)
+		if err != nil {
+			return zero, fmt.Errorf("failed to decode base64 data: %w", err)
 		}
 
-		// Create new instance of concrete type
-		newInstance := reflect.New(concreteType)
+		// We could check and error explicitly if T is an interface type.
 
-		// Unmarshal into the concrete type
-		if err := json.Unmarshal(jsonBytes, newInstance.Interface()); err != nil {
-			return *new(T), fmt.Errorf("unmarshaling for type conversion: %w", err)
+		if err := json.Unmarshal(dataBytes, &zero); err != nil {
+			return zero, fmt.Errorf("failed to unmarshal JSON data: %w", err)
 		}
-
-		// Convert to interface type T
-		return newInstance.Elem().Interface().(T), nil
+		return zero, nil
 	}
 
-	var typedResult T
-	if err := json.Unmarshal(jsonBytes, &typedResult); err != nil {
-		return *new(T), fmt.Errorf("unmarshaling for type conversion: %w", err)
+	// For other serializers, just call the decoder and type-assert
+	decoded, err := serializer.Decode(encoded)
+	if err != nil {
+		return zero, err
+	}
+	typedResult, ok := decoded.(T)
+	if !ok {
+		return zero, fmt.Errorf("cannot convert decoded value of type %T to %T", decoded, zero)
 	}
-
 	return typedResult, nil
 }
 

dbos/serialization_test.go

@@ -1042,18 +1042,15 @@ func TestSerializer(t *testing.T) {
 				isJSON := isJSONSerializer(dbosCtx.serializer)
 
 				if isJSON {
-					// JSON serializer returns map[string]any
-					inputMap, ok := workflow.Input.(map[string]any)
-					require.True(t, ok, "Input should be map[string]any for JSON")
-					assert.Equal(t, input.Message, inputMap["Message"], "Message should match in input")
-					assert.Equal(t, float64(input.Value), inputMap["Value"], "Value should match in input")
-
-					outputMap, ok := workflow.Output.(map[string]any)
-					require.True(t, ok, "Output should be map[string]any for JSON")
-					assert.Equal(t, input.Message, outputMap["Message"], "Message should match in output")
-					assert.Equal(t, float64(input.Value), outputMap["Value"], "Value should match in output")
+					// JSON serializer returns map[string]any. We need to convert it to a _concrete_ type
+					inputConcrete, err := convertJSONToType[ConcreteDataProvider](workflow.Input)
+					require.NoError(t, err, "Failed to convert workflow input to ConcreteDataProvider")
+					assert.Equal(t, input, inputConcrete, "Workflow input should match input")
+
+					outputConcrete, err := convertJSONToType[ConcreteDataProvider](workflow.Output)
+					require.NoError(t, err, "Failed to convert workflow output to ConcreteDataProvider")
+					assert.Equal(t, input, outputConcrete, "Workflow output should match input")
 				} else {
-					// Gob serializer preserves the concrete type
 					inputConcrete, ok := workflow.Input.(ConcreteDataProvider)
 					require.True(t, ok, "Input should be ConcreteDataProvider for Gob")
 					assert.Equal(t, input, inputConcrete, "Input should match")
@@ -1062,6 +1059,46 @@ func TestSerializer(t *testing.T) {
 					require.True(t, ok, "Output should be ConcreteDataProvider for Gob")
 					assert.Equal(t, input, outputConcrete, "Output should match")
 				}
+
+				// Test GetWorkflowSteps for interface types
+				t.Run("GetWorkflowSteps", func(t *testing.T) {
+					steps, err := GetWorkflowSteps(executor, handle.GetWorkflowID())
+					require.NoError(t, err, "Failed to get workflow steps")
+					require.Len(t, steps, 1, "Expected 1 step")
+
+					step := steps[0]
+					require.NotNil(t, step.Output, "Step output should not be nil")
+					assert.Nil(t, step.Error, "Step should not have error")
+
+					if isJSON {
+						// JSON serializer returns map[string]any, convert to ConcreteDataProvider
+						outputConcrete, err := convertJSONToType[ConcreteDataProvider](step.Output)
+						require.NoError(t, err, "Failed to convert step output to ConcreteDataProvider")
+						assert.Equal(t, input, outputConcrete, "Step output should match input")
+					} else {
+						outputConcrete, ok := step.Output.(ConcreteDataProvider)
+						require.True(t, ok, "Output should be ConcreteDataProvider for Gob")
+						assert.Equal(t, input, outputConcrete, "Step output should match input")
+					}
+				})
+
+				// Test RetrieveWorkflow for interface types
+				// TODO: not supported for interface types w/o storing the type information in the DB
+				/*
+					t.Run("RetrieveWorkflow", func(t *testing.T) {
+						h2, err := RetrieveWorkflow[DataProvider](executor, handle.GetWorkflowID())
+						require.NoError(t, err, "Failed to retrieve workflow")
+
+						retrievedResult, err := h2.GetResult()
+						require.NoError(t, err, "Failed to get retrieved workflow result")
+
+						// For interface types, we need to check the concrete type
+						concreteRetrievedResult, ok := retrievedResult.(ConcreteDataProvider)
+						require.True(t, ok, "Retrieved result should be ConcreteDataProvider type")
+						assert.Equal(t, input.Message, concreteRetrievedResult.Message, "Message should match")
+						assert.Equal(t, input.Value, concreteRetrievedResult.Value, "Value should match")
+					})
+				*/
 			})
 
 			// Test nil values with pointer type workflow

dbos/system_database.go

@@ -1541,27 +1541,11 @@ func (s *sysDB) sleep(ctx context.Context, input sleepInput) (time.Duration, err
 			return 0, fmt.Errorf("no recorded end time for recorded sleep operation")
 		}
 
-		// Deserialize the recorded end time
-		decodedOutput, err := s.serializer.Decode(recordedResult.output)
+		// Decode the recorded end time directly into time.Time
+		// recordedResult.output is an encoded *string
+		endTime, err = deserialize[time.Time](s.serializer, recordedResult.output)
 		if err != nil {
-			return 0, fmt.Errorf("failed to deserialize sleep end time: %w", err)
-		}
-
-		// The output should be a time.Time representing the end time
-		// Because checkOperationExecution returns encoded string, we need to decode it into a time.Time
-		endTimeInterface, ok := decodedOutput.(time.Time)
-		if !ok {
-			// JSON serializer loses type information - convert using convertJSONToType
-			if isJSONSerializer(s.serializer) {
-				endTime, err = convertJSONToType[time.Time](decodedOutput)
-				if err != nil {
-					return 0, fmt.Errorf("failed to convert recorded output to time.Time: %w", err)
-				}
-			} else {
-				return 0, fmt.Errorf("decoded output is not a time.Time: %T", decodedOutput)
-			}
-		} else {
-			endTime = endTimeInterface
+			return 0, fmt.Errorf("failed to decode sleep end time: %w", err)
 		}
 
 		if recordedResult.err != nil { // This should never happen

dbos/utils_test.go

@@ -2,6 +2,7 @@ package dbos
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/url"
 	"os"
@@ -181,3 +182,22 @@ func checkWfStatus(ctx DBOSContext, expectedStatus WorkflowStatusType) (bool, er
 	}
 	return false, nil
 }
+
+// Re-encode the value as JSON and then unmarshal into the target type
+func convertJSONToType[T any](value any) (T, error) {
+	if value == nil {
+		return *new(T), nil
+	}
+
+	jsonBytes, err := json.Marshal(value)
+	if err != nil {
+		return *new(T), fmt.Errorf("marshaling for type conversion: %w", err)
+	}
+
+	var typedResult T
+	if err := json.Unmarshal(jsonBytes, &typedResult); err != nil {
+		return *new(T), fmt.Errorf("unmarshaling for type conversion: %w", err)
+	}
+
+	return typedResult, nil
+}