fix(destregistry): handle publisher creation errors as delivery failures (#828)

When CreatePublisher fails (e.g. invalid GCP credentials), the error
was treated as a PreDeliveryError which caused a nack/retry loop until
messages hit the Pub/Sub DLQ — invisible to the customer.

Now:
- GCP Pub/Sub CreatePublisher returns ErrDestinationPublishAttempt for
  validation and client creation errors
- Registry creates a failed attempt when ResolvePublisher returns
  ErrDestinationPublishAttempt or ErrDestinationValidation
- GCP credential JSON is validated for required 'type' field at
  destination creation time

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: alexluong

internal/destregistry/providers/destgcppubsub/destgcppubsub.go

@@ -53,7 +53,10 @@ func (d *GCPPubSubDestination) Validate(ctx context.Context, destination *models
 func (d *GCPPubSubDestination) CreatePublisher(ctx context.Context, destination *models.Destination) (destregistry.Publisher, error) {
 	cfg, creds, err := d.resolveMetadata(ctx, destination)
 	if err != nil {
-		return nil, err
+		return nil, destregistry.NewErrDestinationPublishAttempt(err, "gcp_pubsub", map[string]interface{}{
+			"error":   "validation_failed",
+			"message": err.Error(),
+		})
 	}
 
 	// Create Pub/Sub client options
@@ -74,7 +77,10 @@ func (d *GCPPubSubDestination) CreatePublisher(ctx context.Context, destination
 	// Create the client
 	client, err := pubsub.NewClient(ctx, cfg.ProjectID, opts...)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create Pub/Sub client: %w", err)
+		return nil, destregistry.NewErrDestinationPublishAttempt(err, "gcp_pubsub", map[string]interface{}{
+			"error":   "client_creation_failed",
+			"message": err.Error(),
+		})
 	}
 
 	// Get the topic
@@ -108,6 +114,15 @@ func (d *GCPPubSubDestination) resolveMetadata(ctx context.Context, destination
 				},
 			})
 		}
+		// Validate required GCP credential fields
+		if _, ok := jsonCheck["type"]; !ok {
+			return nil, nil, destregistry.NewErrDestinationValidation([]destregistry.ValidationErrorDetail{
+				{
+					Field: "credentials.service_account_json",
+					Type:  "missing_type",
+				},
+			})
+		}
 	}
 
 	return &GCPPubSubDestinationConfig{

internal/destregistry/providers/destgcppubsub/destgcppubsub_test.go

@@ -179,6 +179,30 @@ func TestValidate(t *testing.T) {
 			},
 			wantErr: false, // Emulator doesn't require valid JSON
 		},
+		{
+			name: "missing type field in service_account_json",
+			config: map[string]string{
+				"project_id": "my-project",
+				"topic":      "my-topic",
+			},
+			credentials: map[string]string{
+				"service_account_json": `{"project_id":"my-project","client_email":"test@test.iam.gserviceaccount.com"}`,
+			},
+			wantErr:     true,
+			errContains: "credentials.service_account_json",
+		},
+		{
+			name: "missing type field - but passes with emulator endpoint",
+			config: map[string]string{
+				"project_id": "my-project",
+				"topic":      "my-topic",
+				"endpoint":   "http://localhost:8085",
+			},
+			credentials: map[string]string{
+				"service_account_json": `{"project_id":"my-project"}`,
+			},
+			wantErr: false,
+		},
 		{
 			name: "valid service account JSON with complete structure",
 			config: map[string]string{

internal/destregistry/registry.go

@@ -138,6 +138,30 @@ func (r *registry) ValidateDestination(ctx context.Context, destination *models.
 func (r *registry) PublishEvent(ctx context.Context, destination *models.Destination, event *models.Event) (*models.Attempt, error) {
 	publisher, err := r.ResolvePublisher(ctx, destination)
 	if err != nil {
+		// If the provider already signaled a delivery error, create a failed attempt
+		// so it's visible to the customer (instead of silently nacking into DLQ).
+		var pubErr *ErrDestinationPublishAttempt
+		var valErr *ErrDestinationValidation
+		if errors.As(err, &pubErr) || errors.As(err, &valErr) {
+			attempt := &models.Attempt{
+				ID:              idgen.Attempt(),
+				DestinationID:   destination.ID,
+				DestinationType: destination.Type,
+				EventID:         event.ID,
+				Time:            time.Now(),
+				Status:          "failed",
+				Code:            "ERR",
+				ResponseData:    map[string]interface{}{"error": err.Error()},
+			}
+			if pubErr != nil {
+				return attempt, pubErr
+			}
+			return attempt, &ErrDestinationPublishAttempt{
+				Err:      err,
+				Provider: destination.Type,
+				Data:     map[string]interface{}{"error": "validation_failed", "message": err.Error()},
+			}
+		}
 		return nil, err
 	}
 

internal/destregistry/registry_test.go

@@ -773,6 +773,120 @@ func TestPublishEventCanceled(t *testing.T) {
 	})
 }
 
+func TestPublishEventResolvePublisherError(t *testing.T) {
+	t.Parallel()
+
+	t.Run("should return failed attempt when CreatePublisher returns ErrDestinationPublishAttempt", func(t *testing.T) {
+		t.Parallel()
+
+		registry := destregistry.NewRegistry(&destregistry.Config{}, testutil.CreateTestLogger(t))
+
+		provider := &mockFailingProvider{
+			createErr: destregistry.NewErrDestinationPublishAttempt(
+				errors.New("missing 'type' field in credentials"),
+				"gcp_pubsub",
+				map[string]interface{}{"error": "client_creation_failed", "message": "missing 'type' field in credentials"},
+			),
+		}
+		err := registry.RegisterProvider("gcp_pubsub", provider)
+		require.NoError(t, err)
+
+		destination := &models.Destination{ID: "dest-1", Type: "gcp_pubsub"}
+		event := &models.Event{ID: "evt-1"}
+
+		attempt, err := registry.PublishEvent(context.Background(), destination, event)
+
+		require.Error(t, err)
+		require.NotNil(t, attempt, "should return a failed attempt, not nil")
+		assert.Equal(t, "failed", attempt.Status)
+		assert.Equal(t, "ERR", attempt.Code)
+		assert.Equal(t, "dest-1", attempt.DestinationID)
+		assert.Equal(t, "evt-1", attempt.EventID)
+
+		var pubErr *destregistry.ErrDestinationPublishAttempt
+		require.ErrorAs(t, err, &pubErr)
+		assert.Equal(t, "client_creation_failed", pubErr.Data["error"])
+	})
+
+	t.Run("should return failed attempt when CreatePublisher returns ErrDestinationValidation", func(t *testing.T) {
+		t.Parallel()
+
+		registry := destregistry.NewRegistry(&destregistry.Config{}, testutil.CreateTestLogger(t))
+
+		provider := &mockFailingProvider{
+			createErr: destregistry.NewErrDestinationValidation([]destregistry.ValidationErrorDetail{
+				{Field: "credentials.service_account_json", Type: "missing_type"},
+			}),
+		}
+		err := registry.RegisterProvider("gcp_pubsub", provider)
+		require.NoError(t, err)
+
+		destination := &models.Destination{ID: "dest-2", Type: "gcp_pubsub"}
+		event := &models.Event{ID: "evt-2"}
+
+		attempt, err := registry.PublishEvent(context.Background(), destination, event)
+
+		require.Error(t, err)
+		require.NotNil(t, attempt, "should return a failed attempt, not nil")
+		assert.Equal(t, "failed", attempt.Status)
+		assert.Equal(t, "ERR", attempt.Code)
+
+		var pubErr *destregistry.ErrDestinationPublishAttempt
+		require.ErrorAs(t, err, &pubErr)
+		assert.Equal(t, "validation_failed", pubErr.Data["error"])
+	})
+
+	t.Run("should return nil attempt for unknown errors", func(t *testing.T) {
+		t.Parallel()
+
+		registry := destregistry.NewRegistry(&destregistry.Config{}, testutil.CreateTestLogger(t))
+
+		provider := &mockFailingProvider{
+			createErr: errors.New("unexpected system error"),
+		}
+		err := registry.RegisterProvider("gcp_pubsub", provider)
+		require.NoError(t, err)
+
+		destination := &models.Destination{ID: "dest-3", Type: "gcp_pubsub"}
+		event := &models.Event{ID: "evt-3"}
+
+		attempt, err := registry.PublishEvent(context.Background(), destination, event)
+
+		require.Error(t, err)
+		assert.Nil(t, attempt, "should return nil attempt for unknown errors")
+	})
+}
+
+// mockFailingProvider simulates a provider whose CreatePublisher always fails
+type mockFailingProvider struct {
+	createErr error
+	*destregistry.BaseProvider
+}
+
+func (p *mockFailingProvider) Validate(ctx context.Context, dest *models.Destination) error {
+	return nil
+}
+
+func (p *mockFailingProvider) CreatePublisher(ctx context.Context, dest *models.Destination) (destregistry.Publisher, error) {
+	return nil, p.createErr
+}
+
+func (p *mockFailingProvider) ComputeTarget(dest *models.Destination) destregistry.DestinationTarget {
+	return destregistry.DestinationTarget{}
+}
+
+func (p *mockFailingProvider) Metadata() *metadata.ProviderMetadata {
+	return &metadata.ProviderMetadata{Type: "gcp_pubsub"}
+}
+
+func (p *mockFailingProvider) ObfuscateDestination(dest *models.Destination) *models.Destination {
+	return dest
+}
+
+func (p *mockFailingProvider) Preprocess(newDest *models.Destination, origDest *models.Destination, opts *destregistry.PreprocessDestinationOpts) error {
+	return nil
+}
+
 func TestDisplayDestination(t *testing.T) {
 	t.Parallel()