[dotnet-port-fixes] Prevent workflow telemetry serialization panics

workflow/inproc/observability_test.go

@@ -24,6 +24,14 @@ func (unserializablePayload) MarshalJSON() ([]byte, error) {
 	return nil, errors.New("marshal failed")
 }
 
+type panicPayload struct {
+	Value string
+}
+
+func (panicPayload) MarshalJSON() ([]byte, error) {
+	panic("marshal panic")
+}
+
 func TestObservability_CreatesWorkflowEndToEndSpans(t *testing.T) {
 	tracer := workflowtest.NewRecordingTracer()
 	wf := newTelemetryWorkflow(t, tracer, workflow.TelemetryOptions{})
@@ -248,6 +256,49 @@ func TestObservability_UnserializableSensitiveDataDoesNotFailWorkflow(t *testing
 	sinkSpan.RequireAttributeValue(t, internalobservability.TagExecutorInput, wantFallback)
 }
 
+func TestObservability_PanickingSensitiveDataDoesNotFailWorkflow(t *testing.T) {
+	tracer := workflowtest.NewRecordingTracer()
+	var received []panicPayload
+	start := workflow.NewExecutor("start", func(input string) panicPayload {
+		return panicPayload{Value: strings.ToUpper(input)}
+	}).Bind()
+	sink := workflow.NewExecutor("sink", func(msg panicPayload) string {
+		received = append(received, msg)
+		return "done"
+	}).Bind()
+
+	wf, err := workflow.NewBuilder(start).
+		AddEdge(start, sink).
+		WithOutputFrom(sink).
+		WithTelemetry(tracer, workflow.TelemetryOptions{EnableSensitiveData: true}).
+		Build()
+	if err != nil {
+		t.Fatalf("Build: %v", err)
+	}
+
+	run, err := inproc.Default.Run(context.Background(), wf, "hello")
+	if err != nil {
+		t.Fatalf("Run: %v", err)
+	}
+	outputs := collectOutputValues(run.OutgoingEvents())
+	if err := run.Close(context.Background()); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+
+	if len(outputs) != 1 || outputs[0] != "done" {
+		t.Fatalf("outputs = %#v, want []string{\"done\"}", outputs)
+	}
+	if len(received) != 1 || received[0].Value != "HELLO" {
+		t.Fatalf("received = %#v, want one delivered payload with value HELLO", received)
+	}
+
+	wantFallback := fmt.Sprintf("[Unserializable: %T]", panicPayload{})
+	messageSpan := workflowtest.FindSpanWithPrefix(t, tracer.Spans(), "message.send")
+	messageSpan.RequireAttributeValue(t, internalobservability.TagMessageContent, wantFallback)
+	sinkSpan := workflowtest.FindSpanWithPrefix(t, tracer.Spans(), "executor.process sink")
+	sinkSpan.RequireAttributeValue(t, internalobservability.TagExecutorInput, wantFallback)
+}
+
 func TestObservability_RunSpansAreEnded(t *testing.T) {
 	tests := []struct {
 		name string

workflow/internal/observability/observability.go

@@ -317,17 +317,26 @@ func setWorkflowAttributes(span *Activity, metadata WorkflowMetadata) {
 	span.SetAttributes(attrs...)
 }
 
-func serialize(value any) string {
+func serialize(value any) (serialized string) {
 	if value == nil {
 		return ""
 	}
+	defer func() {
+		if recover() != nil {
+			serialized = fallbackSerializedValue(value)
+		}
+	}()
 	data, err := json.Marshal(value)
 	if err != nil {
-		return fmt.Sprintf("[Unserializable: %T]", value)
+		return fallbackSerializedValue(value)
 	}
 	return string(data)
 }
 
+func fallbackSerializedValue(value any) string {
+	return fmt.Sprintf("[Unserializable: %T]", value)
+}
+
 type contextKey struct{}
 
 func ContextWithTelemetry(ctx context.Context, telemetry *Context) context.Context {

workflow/internal/observability/observability_test.go

@@ -19,6 +19,12 @@ func (unserializableValue) MarshalJSON() ([]byte, error) {
 	return nil, errors.New("marshal failed")
 }
 
+type panicSerializableValue struct{}
+
+func (panicSerializableValue) MarshalJSON() ([]byte, error) {
+	panic("marshal panic")
+}
+
 func attributeValue(t *testing.T, attrs []workflowobservability.Attribute, key string) string {
 	t.Helper()
 	for _, attr := range attrs {
@@ -131,6 +137,18 @@ func TestSerializedAttributeUsesFallbackForMarshalErrors(t *testing.T) {
 	}
 }
 
+func TestSerializedAttributeUsesFallbackForMarshalPanics(t *testing.T) {
+	attr := observability.SerializedAttribute("message.content", panicSerializableValue{})
+	value, ok := attr.Value.(string)
+	if !ok {
+		t.Fatalf("attribute value type = %T, want string", attr.Value)
+	}
+	want := "[Unserializable: observability_test.panicSerializableValue]"
+	if value != want {
+		t.Fatalf("attribute value = %q, want %q", value, want)
+	}
+}
+
 func TestSensitiveDataUsesFallbackForExecutorInputAndOutput(t *testing.T) {
 	span := &fakeSpan{}
 	telemetry := observability.New(observability.Options{
@@ -153,3 +171,26 @@ func TestSensitiveDataUsesFallbackForExecutorInputAndOutput(t *testing.T) {
 		t.Fatalf("executor.output = %q, want %q", got, want)
 	}
 }
+
+func TestSensitiveDataUsesFallbackForExecutorInputAndOutputPanics(t *testing.T) {
+	span := &fakeSpan{}
+	telemetry := observability.New(observability.Options{
+		Tracer:              &fakeTracer{span: span},
+		EnableSensitiveData: true,
+	})
+
+	message := panicSerializableValue{}
+	_, activity := telemetry.StartExecutorProcess(context.Background(), "exec1", "pkg.Type", "message", message, nil)
+	if activity == nil {
+		t.Fatal("expected an activity span")
+	}
+	telemetry.SetExecutorOutput(activity, message)
+
+	want := "[Unserializable: observability_test.panicSerializableValue]"
+	if got := attributeValue(t, span.attrs, observability.TagExecutorInput); got != want {
+		t.Fatalf("executor.input = %q, want %q", got, want)
+	}
+	if got := attributeValue(t, span.attrs, observability.TagExecutorOutput); got != want {
+		t.Fatalf("executor.output = %q, want %q", got, want)
+	}
+}