record runtime type errors in the typed erased wrapped workflow function

Author: maxdml

dbos/queues_test.go

@@ -439,7 +439,7 @@ func TestQueueRecovery(t *testing.T) {
 	require.True(t, queueEntriesAreCleanedUp(dbosCtx), "expected queue entries to be cleaned up after global concurrency test")
 }
 
-// TODO: we can update this test to have the same logic than TestWorkerConcurrency
+// Note: we could update this test to have the same logic than TestWorkerConcurrency
 func TestGlobalConcurrency(t *testing.T) {
 	dbosCtx := setupDBOS(t, true, true)
 

dbos/system_database.go

@@ -1402,7 +1402,8 @@ func (s *sysDB) notificationListenerLoop(ctx context.Context) {
 				return
 			}
 
-			// Other errors - log and retry. XXX eventually add exponential backoff + jitter
+			// Other errors - log and retry.
+			// TODO add exponential backoff + jitter
 			s.logger.Error("Error waiting for notification", "error", err)
 			time.Sleep(_DB_CONNECTION_RETRY_DELAY)
 			continue

dbos/workflow.go

@@ -392,24 +392,28 @@ func RegisterWorkflow[P any, R any](ctx DBOSContext, fn GenericWorkflowFunc[P, R
 
 	// Register a type-erased version of the durable workflow for recovery
 	typedErasedWorkflow := WorkflowFunc(func(ctx DBOSContext, input any) (any, error) {
-		// This type check is redundant with the one in the wrapper, but I'd better be safe than sorry
 		typedInput, ok := input.(P)
 		if !ok {
-			// FIXME: we need to record the error in the database here
+			wfID, err := ctx.GetWorkflowID()
+			if err != nil {
+				return nil, fmt.Errorf("failed to get workflow ID: %w", err)
+			}
+			err = ctx.(*dbosContext).systemDB.updateWorkflowOutcome(WithoutCancel(ctx), updateWorkflowOutcomeDBInput{
+				workflowID: wfID,
+				status:     WorkflowStatusError,
+				err:        newWorkflowUnexpectedInputType(fqn, fmt.Sprintf("%T", typedInput), fmt.Sprintf("%T", input)),
+			})
+			if err != nil {
+				return nil, fmt.Errorf("failed to record unexpected input type error: %w", err)
+			}
 			return nil, newWorkflowUnexpectedInputType(fqn, fmt.Sprintf("%T", typedInput), fmt.Sprintf("%T", input))
 		}
 		return fn(ctx, typedInput)
 	})
 
 	typeErasedWrapper := WrappedWorkflowFunc(func(ctx DBOSContext, input any, opts ...WorkflowOption) (WorkflowHandle[any], error) {
-		typedInput, ok := input.(P)
-		if !ok {
-			// FIXME: we need to record the error in the database here
-			return nil, newWorkflowUnexpectedInputType(fqn, fmt.Sprintf("%T", typedInput), fmt.Sprintf("%T", input))
-		}
-
 		opts = append(opts, withWorkflowName(fqn)) // Append the name so ctx.RunAsWorkflow can look it up from the registry to apply registration-time options
-		handle, err := ctx.RunAsWorkflow(ctx, typedErasedWorkflow, typedInput, opts...)
+		handle, err := ctx.RunAsWorkflow(ctx, typedErasedWorkflow, input, opts...)
 		if err != nil {
 			return nil, err
 		}