wip

Author: maxdml

dbos/client.go

@@ -242,15 +242,19 @@ func Enqueue[P any, R any](c Client, queueName, workflowName string, input P, op
 
 	// Register the input and outputs for gob encoding
 	var logger *slog.Logger
+	var serializer Serializer
 	if cl, ok := c.(*client); ok {
 		if ctx, ok := cl.dbosCtx.(*dbosContext); ok {
 			logger = ctx.logger
+			serializer = ctx.serializer
 		}
 	}
 	var typedInput P
-	safeGobRegister(typedInput, logger)
 	var typedOutput R
-	safeGobRegister(typedOutput, logger)
+	if isGobSerializer(serializer) {
+		safeGobRegister(typedInput, logger)
+		safeGobRegister(typedOutput, logger)
+	}
 
 	// Call the interface method with the same signature
 	handle, err := c.Enqueue(queueName, workflowName, input, opts...)

dbos/dbos.go

@@ -78,7 +78,7 @@ func processConfig(inputConfig *Config) (*Config, error) {
 		dbosConfig.DatabaseSchema = _DEFAULT_SYSTEM_DB_SCHEMA
 	}
 	if dbosConfig.Serializer == nil {
-		dbosConfig.Serializer = NewJSONSerializer()
+		dbosConfig.Serializer = NewGobSerializer()
 	}
 
 	// Override with environment variables if set
@@ -213,6 +213,7 @@ func WithValue(ctx DBOSContext, key, val any) DBOSContext {
 			applicationVersion:      dbosCtx.applicationVersion,
 			executorID:              dbosCtx.executorID,
 			applicationID:           dbosCtx.applicationID,
+			serializer:              dbosCtx.serializer,
 		}
 		childCtx.launched.Store(launched)
 		return childCtx
@@ -241,6 +242,7 @@ func WithoutCancel(ctx DBOSContext) DBOSContext {
 			applicationVersion:      dbosCtx.applicationVersion,
 			executorID:              dbosCtx.executorID,
 			applicationID:           dbosCtx.applicationID,
+			serializer:              dbosCtx.serializer,
 		}
 		childCtx.launched.Store(launched)
 		return childCtx
@@ -268,6 +270,7 @@ func WithTimeout(ctx DBOSContext, timeout time.Duration) (DBOSContext, context.C
 			applicationVersion:      dbosCtx.applicationVersion,
 			executorID:              dbosCtx.executorID,
 			applicationID:           dbosCtx.applicationID,
+			serializer:              dbosCtx.serializer,
 		}
 		childCtx.launched.Store(launched)
 		return childCtx, cancelFunc
@@ -362,19 +365,21 @@ 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
 	initExecutor.serializer = config.Serializer
 
+	// Register types we serialize with gob (only if using GobSerializer)
+	if isGobSerializer(initExecutor.serializer) {
+		var t time.Time
+		safeGobRegister(t, initExecutor.logger)
+		var ws []WorkflowStatus
+		safeGobRegister(ws, initExecutor.logger)
+		var si []StepInfo
+		safeGobRegister(si, initExecutor.logger)
+	}
+
 	initExecutor.applicationID = os.Getenv("DBOS__APPID")
 
 	newSystemDatabaseInputs := newSystemDatabaseInput{

dbos/errors.go

@@ -55,6 +55,15 @@ func (e *DBOSError) Unwrap() error {
 	return e.wrappedErr
 }
 
+func (e *DBOSError) Is(target error) bool {
+	t, ok := target.(*DBOSError)
+	if !ok {
+		return false
+	}
+	// Match if codes are equal (and target code is set)
+	return t.Code != 0 && e.Code == t.Code
+}
+
 func newConflictingWorkflowError(workflowID, message string) *DBOSError {
 	msg := fmt.Sprintf("Conflicting workflow invocation with the same ID (%s)", workflowID)
 	if message != "" {

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 {

dbos/serialization.go

@@ -33,6 +33,7 @@ func (g *GobSerializer) Encode(data any) (string, error) {
 	var inputBytes []byte
 	if !isNilValue(data) {
 		// Register the type with gob for proper serialization
+		// This is useful when dealing with interface types that didn't have a concrete value at workflow registration time
 		safeGobRegister(data, nil)
 
 		var buf bytes.Buffer
@@ -116,6 +117,42 @@ func serialize(ctx DBOSContext, data any) (string, error) {
 	return dbosCtx.serializer.Encode(data)
 }
 
+// isJSONSerializer checks if the given serializer is a JSONSerializer
+func isJSONSerializer(s Serializer) bool {
+	_, ok := s.(*JSONSerializer)
+	return ok
+}
+
+// isGobSerializer checks if the given serializer is a GobSerializer
+func isGobSerializer(s Serializer) bool {
+	_, ok := s.(*GobSerializer)
+	return ok
+}
+
+// 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.
+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
+}
+
 // Handle cases where the provided data interface wraps a nil value (e.g., var p *int; data := any(p). data != nil but the underlying value is nil)
 func isNilValue(data any) bool {
 	if data == nil {