From e01eaae0fc29515abf00d692895c27dc0f58900b Mon Sep 17 00:00:00 2001 From: Maxence Maireaux Date: Thu, 11 Jun 2026 10:37:15 +0200 Subject: [PATCH] fix(workflow): return errors instead of panicking in history readers ReadInstanceHistory and ReadStageHistory ran inside HTTP handlers but panicked on json.Unmarshal failures and on any DescribeWorkflowExecution error other than NotFound, and indexed Input.Payloads[0] / Result.Payloads[0] without a length check (index out of range when a workflow/activity was started with no payload). chi's Recoverer turned each into an opaque 500. - Add unmarshalFirstPayload(), which tolerates a nil/empty payload set and returns the decode error instead of panicking. - Replace all four panic sites with wrapped returned errors; the DescribeWorkflowExecution NotFound case still maps to ErrInstanceNotFound. Adds TestUnmarshalFirstPayload (nil/empty/malformed/well-formed). Note: overlaps internal/workflow/manager.go with other PRs in this series (different functions); independent. --- internal/workflow/manager.go | 25 ++++++++++++++++++------- internal/workflow/manager_test.go | 22 ++++++++++++++++++++++ 2 files changed, 40 insertions(+), 7 deletions(-) diff --git a/internal/workflow/manager.go b/internal/workflow/manager.go index 8460c61..11b5e40 100644 --- a/internal/workflow/manager.go +++ b/internal/workflow/manager.go @@ -9,6 +9,7 @@ import ( "github.com/formancehq/go-libs/v3/pointer" + common "go.temporal.io/api/common/v1" enums "go.temporal.io/api/enums/v1" history "go.temporal.io/api/history/v1" @@ -209,6 +210,16 @@ type StageHistory struct { TerminatedAt *time.Time `json:"terminatedAt,omitempty"` } +// unmarshalFirstPayload decodes the first Temporal payload into v. It tolerates +// a nil/empty payload set (leaving v untouched) instead of panicking on an +// out-of-range index, and returns the decode error rather than panicking. +func unmarshalFirstPayload(payloads *common.Payloads, v any) error { + if payloads == nil || len(payloads.Payloads) == 0 { + return nil + } + return json.Unmarshal(payloads.Payloads[0].Data, v) +} + func (m *WorkflowManager) ReadInstanceHistory(ctx context.Context, instanceID string) ([]StageHistory, error) { historyIterator := m.temporalClient.GetWorkflowHistory(ctx, instanceID+"-main", "", @@ -223,8 +234,8 @@ func (m *WorkflowManager) ReadInstanceHistory(ctx context.Context, instanceID st case enums.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED: attributes := event.Attributes.(*history.HistoryEvent_StartChildWorkflowExecutionInitiatedEventAttributes) input := make(map[string]any) - if err := json.Unmarshal(attributes.StartChildWorkflowExecutionInitiatedEventAttributes.Input.Payloads[0].Data, &input); err != nil { - panic(err) + if err := unmarshalFirstPayload(attributes.StartChildWorkflowExecutionInitiatedEventAttributes.Input, &input); err != nil { + return nil, errors.Wrap(err, "unmarshalling stage input") } stageHistory := StageHistory{ Name: attributes.StartChildWorkflowExecutionInitiatedEventAttributes.WorkflowType.Name, @@ -281,7 +292,7 @@ func (m *WorkflowManager) ReadStageHistory(ctx context.Context, instanceID strin if _, ok := err.(*serviceerror.NotFound); ok { return nil, ErrInstanceNotFound } - panic(err) + return nil, errors.Wrap(err, "describing workflow execution") } historyIterator := m.temporalClient.GetWorkflowHistory(ctx, stageID, "", @@ -296,8 +307,8 @@ func (m *WorkflowManager) ReadStageHistory(ctx context.Context, instanceID strin case enums.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED: activityTaskScheduledEventAttributes := event.Attributes.(*history.HistoryEvent_ActivityTaskScheduledEventAttributes).ActivityTaskScheduledEventAttributes input := make(map[string]any) - if err := json.Unmarshal(activityTaskScheduledEventAttributes.Input.Payloads[0].Data, &input); err != nil { - panic(err) + if err := unmarshalFirstPayload(activityTaskScheduledEventAttributes.Input, &input); err != nil { + return nil, errors.Wrap(err, "unmarshalling activity input") } activityHistory := &ActivityHistory{ @@ -334,8 +345,8 @@ func (m *WorkflowManager) ReadStageHistory(ctx context.Context, instanceID strin result := event.Attributes.(*history.HistoryEvent_ActivityTaskCompletedEventAttributes).ActivityTaskCompletedEventAttributes.Result if result != nil && len(result.Payloads) > 0 { output := make(map[string]any) - if err := json.Unmarshal(result.Payloads[0].Data, &output); err != nil { - panic(err) + if err := unmarshalFirstPayload(result, &output); err != nil { + return nil, errors.Wrap(err, "unmarshalling activity output") } // notes(gfyrag): keep compat with format from ledger v1 (since we have moved to ledger v2 api) diff --git a/internal/workflow/manager_test.go b/internal/workflow/manager_test.go index 1ce2646..6c9310e 100644 --- a/internal/workflow/manager_test.go +++ b/internal/workflow/manager_test.go @@ -18,8 +18,30 @@ import ( "github.com/formancehq/orchestration/internal/storage" "github.com/stretchr/testify/require" + common "go.temporal.io/api/common/v1" ) +func TestUnmarshalFirstPayload(t *testing.T) { + t.Parallel() + + var v map[string]any + // nil and empty payload sets are tolerated (no panic, no error). + require.NoError(t, unmarshalFirstPayload(nil, &v)) + require.NoError(t, unmarshalFirstPayload(&common.Payloads{}, &v)) + + // A malformed payload returns an error instead of panicking. + err := unmarshalFirstPayload(&common.Payloads{ + Payloads: []*common.Payload{{Data: []byte("{not-json")}}, + }, &v) + require.Error(t, err) + + // A well-formed payload decodes. + require.NoError(t, unmarshalFirstPayload(&common.Payloads{ + Payloads: []*common.Payload{{Data: []byte(`{"a":"b"}`)}}, + }, &v)) + require.Equal(t, map[string]any{"a": "b"}, v) +} + func TestConfig(t *testing.T) { t.Parallel()