aggregatable error message (#21767)

* aggregatable error message

* lint

Author: patrickhuie19

core/services/workflows/artifacts/v2/store.go

@@ -203,7 +203,7 @@ func (h *Store) FetchWorkflowArtifacts(ctx context.Context, workflowID, binaryUR
 	}
 	binary, err = h.fetchFn(ctx, messageID(binaryURL, workflowID), req)
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to fetch binary from %s : %w", binaryURL, err)
+		return nil, nil, &types.ArtifactFetchError{ArtifactType: "binary", URL: binaryURL, Err: err}
 	}
 
 	if decodedBinary, err = base64.StdEncoding.DecodeString(string(binary)); err != nil {
@@ -247,7 +247,7 @@ func (h *Store) FetchWorkflowArtifacts(ctx context.Context, workflowID, binaryUR
 
 		config, err2 = h.fetchFn(ctx, messageID(configURL, workflowID), req)
 		if err2 != nil {
-			return nil, nil, fmt.Errorf("failed to fetch config from %s : %w", configURL, err2)
+			return nil, nil, &types.ArtifactFetchError{ArtifactType: "config", URL: configURL, Err: err2}
 		}
 	}
 	return decodedBinary, config, nil

core/services/workflows/syncer/v2/handler.go

@@ -294,13 +294,14 @@ func (h *eventHandler) Handle(ctx context.Context, event Event) error {
 
 		var err error
 		defer func() {
-			if err2 := events.EmitWorkflowStatusChangedEventV2(ctx, cma.Labels(), toCommonHead(event.Head), string(event.Name), payload.BinaryURL, payload.ConfigURL, err); err2 != nil {
+			if err2 := events.EmitWorkflowStatusChangedEventV2(ctx, cma.Labels(), toCommonHead(event.Head), string(event.Name), payload.BinaryURL, payload.ConfigURL, customerFacingError(err)); err2 != nil {
 				h.lggr.Errorf("failed to emit status changed event: %+v", err2)
 			}
 		}()
 		err = h.workflowActivatedEvent(ctx, payload)
 		if err != nil {
-			logCustMsg(ctx, cma, fmt.Sprintf("failed to handle workflow activated event: %v", err), h.lggr)
+			h.lggr.Errorw("failed to handle workflow activated event", "error", err, "workflowID", wfID)
+			logCustMsg(ctx, cma, fmt.Sprintf("failed to handle workflow activated event: %v", customerFacingError(err)), h.lggr)
 			return err
 		}
 
@@ -888,6 +889,20 @@ func (h *eventHandler) ensureCapRegistryReady(ctx context.Context) error {
 		})
 }
 
+// customerFacingError returns a deterministic, user-actionable error for beholder emission.
+// Internal errors (e.g. ArtifactFetchError with per-node signed URLs) are replaced with a
+// clean message so that workflow-service can aggregate error_message across nodes.
+func customerFacingError(err error) error {
+	if err == nil {
+		return nil
+	}
+	var fetchErr *types.ArtifactFetchError
+	if errors.As(err, &fetchErr) {
+		return errors.New(fetchErr.CustomerError())
+	}
+	return err
+}
+
 func newHandlerTypeError(data any) error {
 	return fmt.Errorf("invalid data type %T for event", data)
 }

core/services/workflows/syncer/v2/handler_test.go

@@ -688,6 +688,40 @@ func testRunningWorkflow(t *testing.T, tc testCase) {
 	})
 }
 
+func Test_customerFacingError(t *testing.T) {
+	t.Run("nil error returns nil", func(t *testing.T) {
+		assert.NoError(t, customerFacingError(nil))
+	})
+
+	t.Run("ArtifactFetchError returns deterministic customer message", func(t *testing.T) {
+		fetchErr := &types.ArtifactFetchError{
+			ArtifactType: "binary",
+			URL:          "https://storage.example.com/binary.wasm?Expires=123&Signature=nodeSpecificSig",
+			Err:          errors.New("connection refused"),
+		}
+		got := customerFacingError(fetchErr)
+		require.Error(t, got)
+		assert.Equal(t, "Internal error: failed to fetch workflow binary from storage. Contact support if this persists.", got.Error())
+	})
+
+	t.Run("wrapped ArtifactFetchError is still detected", func(t *testing.T) {
+		fetchErr := &types.ArtifactFetchError{
+			ArtifactType: "config",
+			URL:          "https://storage.example.com/config.yaml?Expires=456&Signature=abc",
+			Err:          errors.New("timeout"),
+		}
+		wrapped := fmt.Errorf("createWorkflowSpec: %w", fetchErr)
+		got := customerFacingError(wrapped)
+		assert.Contains(t, got.Error(), "workflow config")
+		assert.NotContains(t, got.Error(), "Expires")
+	})
+
+	t.Run("non-ArtifactFetchError passes through unchanged", func(t *testing.T) {
+		original := errors.New("some other error")
+		assert.Equal(t, original, customerFacingError(original))
+	})
+}
+
 type mockArtifactStore struct {
 	artifactStore              *artifacts.Store
 	deleteWorkflowArtifactsErr error

core/services/workflows/types/errors.go

@@ -1,8 +1,32 @@
 package types
 
-import "errors"
+import (
+	"errors"
+	"fmt"
+)
 
 var (
 	ErrGlobalWorkflowCountLimitReached   = errors.New("global workflow count limit reached")
 	ErrPerOwnerWorkflowCountLimitReached = errors.New("per owner workflow count limit reached")
 )
+
+// ArtifactFetchError represents an internal failure to fetch a workflow artifact.
+// It preserves full details for developer debugging while providing a deterministic
+// customer-facing message suitable for aggregation across nodes.
+type ArtifactFetchError struct {
+	ArtifactType string // "binary" or "config"
+	URL          string
+	Err          error
+}
+
+func (e *ArtifactFetchError) Error() string {
+	return fmt.Sprintf("failed to fetch %s from %s : %s", e.ArtifactType, e.URL, e.Err)
+}
+
+func (e *ArtifactFetchError) Unwrap() error {
+	return e.Err
+}
+
+func (e *ArtifactFetchError) CustomerError() string {
+	return fmt.Sprintf("Internal error: failed to fetch workflow %s from storage. Contact support if this persists.", e.ArtifactType)
+}

core/services/workflows/types/errors_test.go

@@ -0,0 +1,49 @@
+package types
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestArtifactFetchError(t *testing.T) {
+	inner := errors.New("connection refused")
+	fetchErr := &ArtifactFetchError{
+		ArtifactType: "binary",
+		URL:          "https://storage.example.com/artifacts/abc123/binary.wasm?Expires=123&Signature=xyz",
+		Err:          inner,
+	}
+
+	t.Run("Error preserves full URL for internal debugging", func(t *testing.T) {
+		assert.Contains(t, fetchErr.Error(), "Expires=123&Signature=xyz")
+		assert.Contains(t, fetchErr.Error(), "binary.wasm")
+		assert.Contains(t, fetchErr.Error(), "connection refused")
+	})
+
+	t.Run("CustomerError is deterministic and omits URL details", func(t *testing.T) {
+		msg := fetchErr.CustomerError()
+		assert.Equal(t, "Internal error: failed to fetch workflow binary from storage. Contact support if this persists.", msg)
+		assert.NotContains(t, msg, "Expires")
+		assert.NotContains(t, msg, "Signature")
+		assert.NotContains(t, msg, "example.com")
+	})
+
+	t.Run("CustomerError reflects artifact type", func(t *testing.T) {
+		configErr := &ArtifactFetchError{ArtifactType: "config", URL: "https://x.com/c?s=1", Err: inner}
+		assert.Contains(t, configErr.CustomerError(), "workflow config")
+	})
+
+	t.Run("Unwrap returns inner error", func(t *testing.T) {
+		require.ErrorIs(t, fetchErr, inner)
+	})
+
+	t.Run("errors.As matches through wrapping", func(t *testing.T) {
+		wrapped := fmt.Errorf("outer: %w", fetchErr)
+		var target *ArtifactFetchError
+		require.ErrorAs(t, wrapped, &target)
+		assert.Equal(t, "binary", target.ArtifactType)
+	})
+}