Json serializer (#183)

- use JSON for encoding/decoding
- lift encode/decode out of the system database
- distinguish nil values (stored as NULL) in Postgres
- return json strings on the ListWorkflows GetWorkflowSteps paths
- Add comprehensive testing suite

On the get wf steps / list wf path, the contract is that if the value is
nil, you know it was storing a nil value. Else, the value is a JSON
string of your workflow's or step's `T`.

Author: maxdml

dbos/admin_server.go

@@ -152,20 +152,23 @@ func toListWorkflowResponse(ws WorkflowStatus) (map[string]any, error) {
 		result["StartedAt"] = nil
 	}
 
-	if ws.Input != nil && ws.Input != "" {
-		bytes, err := json.Marshal(ws.Input)
-		if err != nil {
-			return nil, fmt.Errorf("failed to marshal input: %w", err)
+	if ws.Input != nil {
+		// If there is a value, it should be a JSON string
+		jsonInput, ok := ws.Input.(string)
+		if ok {
+			result["Input"] = jsonInput
+		} else {
+			result["Input"] = ""
 		}
-		result["Input"] = string(bytes)
 	}
 
-	if ws.Output != nil && ws.Output != "" {
-		bytes, err := json.Marshal(ws.Output)
-		if err != nil {
-			return nil, fmt.Errorf("failed to marshal output: %w", err)
+	if ws.Output != nil {
+		jsonOutput, ok := ws.Output.(string)
+		if ok {
+			result["Output"] = jsonOutput
+		} else {
+			result["Output"] = ""
 		}
-		result["Output"] = string(bytes)
 	}
 
 	if ws.Error != nil {
@@ -439,15 +442,16 @@ func newAdminServer(ctx *dbosContext, port int) *adminServer {
 				"child_workflow_id": step.ChildWorkflowID,
 			}
 
-			// Marshal Output as JSON string if present
-			if step.Output != nil && step.Output != "" {
-				bytes, err := json.Marshal(step.Output)
-				if err != nil {
-					ctx.logger.Error("Failed to marshal step output", "error", err)
-					http.Error(w, fmt.Sprintf("Failed to format step output: %v", err), http.StatusInternalServerError)
-					return
+			if step.Output != nil {
+				// If there is a value, it should be a JSON string
+				jsonOutput, ok := step.Output.(string)
+				if ok {
+					formattedStep["output"] = jsonOutput
+				} else {
+					formattedStep["output"] = ""
 				}
-				formattedStep["output"] = string(bytes)
+			} else {
+				formattedStep["output"] = ""
 			}
 
 			// Marshal Error as JSON string if present

dbos/admin_server_test.go

@@ -347,12 +347,12 @@ func TestAdminServer(t *testing.T) {
 				// Empty string workflow: both input and output are empty strings
 				// According to the logic, empty strings should not have Input/Output fields
 				input, hasInput := wf["Input"]
-				require.Equal(t, "", input)
+				require.Equal(t, "\"\"", input)
 				require.True(t, hasInput, "Empty string workflow should have Input field")
 
 				output, hasOutput := wf["Output"]
 				require.True(t, hasOutput, "Empty string workflow should have Output field")
-				require.Equal(t, "", output)
+				require.Equal(t, "\"\"", output)
 
 			} else if wfID == structHandle.GetWorkflowID() {
 				// Struct workflow: input and output should be marshaled as JSON strings
@@ -874,22 +874,9 @@ func TestAdminServer(t *testing.T) {
 				assert.Contains(t, unmarshaledError, "deliberate error for testing", "Error message should be preserved")
 
 			case "emptyStep":
-				// Empty string might be returned as nil or as an empty JSON string
+				// Empty string is returned as an empty JSON string
 				output := step["output"]
-				if output == nil {
-					// Empty string was not included in response (which is fine)
-					t.Logf("Empty step output was nil (not included)")
-				} else {
-					// If it was included, it should be marshaled as JSON string `""`
-					outputStr, ok := output.(string)
-					require.True(t, ok, "If present, empty step output should be a JSON string")
-
-					var unmarshaledOutput string
-					err = json.Unmarshal([]byte(outputStr), &unmarshaledOutput)
-					require.NoError(t, err, "Failed to unmarshal empty step output")
-					assert.Equal(t, "", unmarshaledOutput, "Empty step output should be empty string")
-				}
-
+				require.Equal(t, "\"\"", output, "Empty step output should be an empty string")
 				assert.Nil(t, step["error"], "Empty step should have no error")
 			}
 		}

dbos/client.go

@@ -147,6 +147,7 @@ func (c *client) Enqueue(queueName, workflowName string, input any, opts ...Enqu
 	if params.priority > uint(math.MaxInt) {
 		return nil, fmt.Errorf("priority %d exceeds maximum allowed value %d", params.priority, math.MaxInt)
 	}
+
 	status := WorkflowStatus{
 		Name:               params.workflowName,
 		ApplicationVersion: params.applicationVersion,
@@ -155,7 +156,7 @@ func (c *client) Enqueue(queueName, workflowName string, input any, opts ...Enqu
 		CreatedAt:          time.Now(),
 		Deadline:           deadline,
 		Timeout:            params.workflowTimeout,
-		Input:              params.workflowInput,
+		Input:              input,
 		QueueName:          queueName,
 		DeduplicationID:    params.deduplicationID,
 		Priority:           int(params.priority),
@@ -240,20 +241,15 @@ func Enqueue[P any, R any](c Client, queueName, workflowName string, input P, op
 		return nil, errors.New("client cannot be nil")
 	}
 
-	// Register the input and outputs for gob encoding
-	var logger *slog.Logger
-	if cl, ok := c.(*client); ok {
-		if ctx, ok := cl.dbosCtx.(*dbosContext); ok {
-			logger = ctx.logger
-		}
+	// Serialize input
+	serializer := newJSONSerializer[P]()
+	encodedInput, err := serializer.Encode(input)
+	if err != nil {
+		return nil, fmt.Errorf("failed to serialize workflow input: %w", err)
 	}
-	var typedInput P
-	safeGobRegister(typedInput, logger)
-	var typedOutput R
-	safeGobRegister(typedOutput, logger)
 
 	// Call the interface method with the same signature
-	handle, err := c.Enqueue(queueName, workflowName, input, opts...)
+	handle, err := c.Enqueue(queueName, workflowName, encodedInput, opts...)
 	if err != nil {
 		return nil, err
 	}

dbos/client_test.go

@@ -369,11 +369,11 @@ func TestCancelResume(t *testing.T) {
 
 		// Wait for workflow completion
 		proceedSignal.Set() // Allow the workflow to proceed to step two
-		result, err := resumeHandle.GetResult()
+		resultAny, err := resumeHandle.GetResult()
 		require.NoError(t, err, "failed to get result from resumed workflow")
 
-		// Verify the result
-		assert.Equal(t, input, result, "expected result to match input")
+		// Will be a float64 from json decode
+		require.Equal(t, input, int(resultAny.(float64)), "expected result to match input")
 
 		// Verify both steps completed
 		assert.Equal(t, 2, stepsCompleted, "expected steps completed to be 2")
@@ -391,10 +391,11 @@ func TestCancelResume(t *testing.T) {
 		resumeAgainHandle, err := client.ResumeWorkflow(workflowID)
 		require.NoError(t, err, "failed to resume workflow again")
 
-		resultAgain, err := resumeAgainHandle.GetResult()
+		resultAgainAny, err := resumeAgainHandle.GetResult()
 		require.NoError(t, err, "failed to get result from second resume")
 
-		assert.Equal(t, input, resultAgain, "expected second resume result to match input")
+		// Will be a float64 from json decode
+		require.Equal(t, input, int(resultAgainAny.(float64)), "expected result to match input")
 
 		// Verify steps didn't run again
 		assert.Equal(t, 2, stepsCompleted, "expected steps completed to remain 2 after second resume")

dbos/dbos.go

@@ -185,6 +185,7 @@ func (c *dbosContext) Value(key any) any {
 	return c.ctx.Value(key)
 }
 
+
 // WithValue returns a copy of the DBOS context with the given key-value pair.
 // This is similar to context.WithValue but maintains DBOS context capabilities.
 // No-op if the provided context is not a concrete dbos.dbosContext.
@@ -354,14 +355,6 @@ func NewDBOSContext(ctx context.Context, inputConfig Config) (DBOSContext, error
 	initExecutor.logger = config.Logger
 	initExecutor.logger.Info("Initializing DBOS context", "app_name", config.AppName, "dbos_version", getDBOSVersion())
 
-	// Register types we serialize with gob
-	var t time.Time
-	safeGobRegister(t, initExecutor.logger)
-	var ws []WorkflowStatus
-	safeGobRegister(ws, initExecutor.logger)
-	var si []StepInfo
-	safeGobRegister(si, initExecutor.logger)
-
 	// Initialize global variables from processed config (already handles env vars and defaults)
 	initExecutor.applicationVersion = config.ApplicationVersion
 	initExecutor.executorID = config.ExecutorID

dbos/queue.go

@@ -1,10 +1,7 @@
 package dbos
 
 import (
-	"bytes"
 	"context"
-	"encoding/base64"
-	"encoding/gob"
 	"log/slog"
 	"math"
 	"math/rand"
@@ -227,23 +224,8 @@ 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
-					}
-				}
-
-				_, 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/queues_test.go

@@ -2,6 +2,7 @@ package dbos
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"os"
@@ -538,10 +539,16 @@ func TestQueueRecovery(t *testing.T) {
 	for _, h := range recoveryHandles {
 		if h.GetWorkflowID() == wfid {
 			// Root workflow case
-			result, err := h.GetResult()
+			resultAny, err := h.GetResult()
 			require.NoError(t, err, "failed to get result from recovered root workflow handle")
-			castedResult, ok := result.([]int)
-			require.True(t, ok, "expected result to be of type []int for root workflow, got %T", result)
+			// re-encode and decode the result from []interface{} to []int
+			encodedResult, ok := resultAny.([]any)
+			require.True(t, ok, "expected result to be a []any")
+			jsonBytes, err := json.Marshal(encodedResult)
+			require.NoError(t, err, "failed to marshal result to JSON")
+			var castedResult []int
+			err = json.Unmarshal(jsonBytes, &castedResult)
+			require.NoError(t, err, "failed to decode result to []int")
 			expectedResult := []int{0, 1, 2, 3, 4}
 			assert.Equal(t, expectedResult, castedResult, "expected result %v, got %v", expectedResult, castedResult)
 		}

dbos/recovery.go

@@ -1,9 +1,5 @@
 package dbos
 
-import (
-	"strings"
-)
-
 func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]WorkflowHandle[any], error) {
 	workflowHandles := make([]WorkflowHandle[any], 0)
 	// List pending workflows for the executors
@@ -18,13 +14,6 @@ func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]Workflow
 	}
 
 	for _, workflow := range pendingWorkflows {
-		if inputStr, ok := workflow.Input.(string); ok {
-			if strings.Contains(inputStr, "Failed to decode") {
-				ctx.logger.Warn("Skipping workflow recovery due to input decoding failure", "workflow_id", workflow.ID, "name", workflow.Name)
-				continue
-			}
-		}
-
 		if workflow.QueueName != "" {
 			cleared, err := ctx.systemDB.clearQueueAssignment(ctx, workflow.ID)
 			if err != nil {
@@ -59,6 +48,7 @@ func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]Workflow
 			WithWorkflowID(workflow.ID),
 		}
 		// Create a workflow context from the executor context
+		// Pass encoded input directly - decoding will happen in workflow wrapper when we know the target type
 		handle, err := registeredWorkflow.wrappedFunction(ctx, workflow.Input, opts...)
 		if err != nil {
 			return nil, err