fix(destregistry): treat event format errors as failed deliveries, not DLQ

cmd/e2e/regression_format_error_test.go

@@ -0,0 +1,176 @@
+package e2e_test
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/gin-gonic/gin"
+	"github.com/hookdeck/outpost/cmd/e2e/configs"
+	"github.com/hookdeck/outpost/internal/app"
+	"github.com/hookdeck/outpost/internal/config"
+	"github.com/hookdeck/outpost/internal/util/testinfra"
+	"github.com/stretchr/testify/require"
+)
+
+// TestE2E_Regression_FormatErrorIsDeliveredAttempt verifies that a per-event
+// formatting failure is handled as a normal failed delivery, not a system error
+// that gets nacked and dead-lettered.
+//
+// We use aws_s3 because its key_template is only syntax-validated at creation, so a
+// valid template can still fail at delivery for an event missing a referenced field.
+//
+// Asserts, end to end:
+//  1. nothing is written to S3 (the failure is pre-delivery)
+//  2. each attempt is recorded as a failed delivery
+//  3. it retries on the normal schedule and exhausts, rather than being dead-lettered
+func TestE2E_Regression_FormatErrorIsDeliveredAttempt(t *testing.T) {
+	t.Parallel()
+	if testing.Short() {
+		t.Skip("skipping e2e test")
+	}
+
+	testinfraCleanup := testinfra.Start(t)
+	defer testinfraCleanup()
+	gin.SetMode(gin.TestMode)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// LocalStack S3: the bucket doubles as the "was anything actually delivered?" sink.
+	endpoint := testinfra.EnsureLocalStack()
+	awsCfg, err := awsconfig.LoadDefaultConfig(ctx,
+		awsconfig.WithRegion("us-east-1"),
+		awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
+	)
+	require.NoError(t, err)
+	s3Client := s3.NewFromConfig(awsCfg, func(o *s3.Options) {
+		o.UsePathStyle = true // required for LocalStack
+		o.BaseEndpoint = aws.String(endpoint)
+	})
+	bucket := fmt.Sprintf("regr-format-%d", time.Now().UnixNano())
+	_, err = s3Client.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String(bucket)})
+	require.NoError(t, err)
+
+	// Isolated outpost instance with a short, bounded retry schedule so the budget
+	// exhausts quickly: schedule length 2 => 2 retries => 3 total attempts.
+	cfg := configs.Basic(t, configs.BasicOpts{LogStorage: configs.LogStorageTypeClickHouse})
+	cfg.RetrySchedule = []int{1, 1}
+	cfg.RetryPollBackoffMs = 50
+	cfg.LogBatchThresholdSeconds = 0 // immediate flush so /attempts is reliable
+	require.NoError(t, cfg.Validate(config.Flags{}))
+	configs.ApplyMigrations(t, &cfg)
+
+	appDone := make(chan struct{})
+	go func() {
+		defer close(appDone)
+		application := app.New(&cfg)
+		if err := application.Run(ctx); err != nil {
+			log.Println("Application stopped:", err)
+		}
+	}()
+	defer func() {
+		cancel()
+		<-appDone
+	}()
+
+	waitForHealthy(t, cfg.APIPort, 5*time.Second)
+
+	client := newRegressionHTTPClient(cfg.APIKey)
+	apiURL := fmt.Sprintf("http://localhost:%d/api/v1", cfg.APIPort)
+
+	tenantID := fmt.Sprintf("tenant_format_%d", time.Now().UnixNano())
+	destinationID := fmt.Sprintf("dest_format_%d", time.Now().UnixNano())
+	eventID := fmt.Sprintf("evt_format_%d", time.Now().UnixNano())
+
+	// Create tenant.
+	status := client.doJSON(t, http.MethodPut, apiURL+"/tenants/"+tenantID, nil, nil)
+	require.Equal(t, 201, status, "failed to create tenant")
+
+	// Create an aws_s3 destination whose key_template is valid at creation but cannot be
+	// evaluated for an event lacking metadata.operationId (join over a nil value fails).
+	status = client.doJSON(t, http.MethodPost, apiURL+"/tenants/"+tenantID+"/destinations", map[string]any{
+		"id":     destinationID,
+		"type":   "aws_s3",
+		"topics": "*",
+		"config": map[string]any{
+			"bucket":        bucket,
+			"region":        "us-east-1",
+			"endpoint":      endpoint,
+			"storage_class": "STANDARD",
+			"key_template":  `join('/', ['prefix', metadata.operationId])`,
+		},
+		"credentials": map[string]any{
+			"key":    "test",
+			"secret": "test",
+		},
+	}, nil)
+	require.Equal(t, 201, status, "failed to create aws_s3 destination")
+
+	// Publish a retry-eligible event WITHOUT metadata.operationId -> formatting fails at delivery.
+	status = client.doJSON(t, http.MethodPost, apiURL+"/publish", map[string]any{
+		"id":                 eventID,
+		"tenant_id":          tenantID,
+		"topic":              "user.created",
+		"eligible_for_retry": true,
+		"metadata":           map[string]any{"foo": "bar"},
+		"data":               map[string]any{"hello": "world"},
+	}, nil)
+	require.Equal(t, 202, status, "failed to publish event")
+
+	attemptsURL := apiURL + "/attempts?tenant_id=" + tenantID + "&event_id=" + eventID + "&dir=asc&include=response_data"
+	pollAttempts := func(t *testing.T, minCount int, timeout time.Duration) []map[string]any {
+		t.Helper()
+		deadline := time.Now().Add(timeout)
+		for time.Now().Before(deadline) {
+			var resp struct {
+				Models []map[string]any `json:"models"`
+			}
+			s := client.doJSON(t, http.MethodGet, attemptsURL, nil, &resp)
+			if s == http.StatusOK && len(resp.Models) >= minCount {
+				return resp.Models
+			}
+			time.Sleep(100 * time.Millisecond)
+		}
+		t.Fatalf("timed out waiting for %d attempts", minCount)
+		return nil
+	}
+
+	// (2) + (3): retries run on the normal schedule and produce recorded failed attempts.
+	// 1 initial + 2 scheduled retries = 3 attempts. If the fix regressed, the format error
+	// would nack/dead-letter the message and ZERO attempts would be logged -> this times out.
+	attempts := pollAttempts(t, 3, 15*time.Second)
+	for i, atm := range attempts {
+		require.Equal(t, "failed", atm["status"], "attempt %d should be a failed delivery", i+1)
+		// (2) the attempt carries the format error as a normal, customer-facing delivery error.
+		if rd, ok := atm["response_data"].(map[string]any); ok {
+			require.Equal(t, "could not format event for delivery", rd["error"],
+				"attempt %d should record the format error", i+1)
+		} else {
+			t.Fatalf("attempt %d missing response_data", i+1)
+		}
+	}
+
+	// Budget exhausted: after the schedule completes, no further attempts appear. A
+	// dead-letter/requeue loop would keep producing attempts (or none at all) instead
+	// of stopping cleanly at the retry budget.
+	time.Sleep(2 * time.Second)
+	var finalResp struct {
+		Models []map[string]any `json:"models"`
+	}
+	client.doJSON(t, http.MethodGet, attemptsURL, nil, &finalResp)
+	require.Len(t, finalResp.Models, 3,
+		"should have exactly 3 attempts (1 initial + 2 retries) then stop — not requeue into a DLQ")
+
+	// (1) nothing was ever written to S3 — the failure happened before any PutObject.
+	listed, err := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(bucket)})
+	require.NoError(t, err)
+	require.Empty(t, listed.Contents, "no object should have been written to S3")
+}

internal/destregistry/error.go

@@ -39,6 +39,27 @@ func NewErrDestinationPublishAttempt(err error, provider string, data map[string
 	return &ErrDestinationPublishAttempt{Err: err, Provider: provider, Data: data}
 }
 
+// NewFormatError returns the (*Delivery, error) a publisher should return when
+// formatting an event fails before it can be sent (e.g. an invalid key/partition
+// template or an unparseable payload). It records a failed attempt so the failure
+// is visible to the customer and the message is acked, instead of nacking into the DLQ.
+//
+// message is the customer-facing string persisted on the attempt (ResponseData);
+// when empty a generic default is used. The raw err is carried only in the returned
+// error (for logs/telemetry) and is not persisted on the attempt.
+func NewFormatError(provider, message string, err error) (*Delivery, error) {
+	if message == "" {
+		message = "could not format event for delivery"
+	}
+	return &Delivery{
+			Status:   "failed",
+			Code:     "ERR",
+			Response: map[string]interface{}{"error": message},
+		}, NewErrDestinationPublishAttempt(err, provider, map[string]interface{}{
+			"error": "format_failed",
+		})
+}
+
 // NewErrPublishCanceled creates an error for when publish is canceled (e.g., service shutdown).
 // This should return nil Delivery to trigger nack → requeue for another instance.
 // See: https://github.com/hookdeck/outpost/issues/571

internal/destregistry/format_error_test.go

@@ -0,0 +1,52 @@
+package destregistry_test
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/hookdeck/outpost/internal/destregistry"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewFormatError(t *testing.T) {
+	t.Parallel()
+
+	rawErr := errors.New("failed to evaluate key template: invalid type")
+
+	t.Run("returns a failed delivery and a publish-attempt error", func(t *testing.T) {
+		t.Parallel()
+
+		delivery, err := destregistry.NewFormatError("aws_s3", "", rawErr)
+
+		// A non-nil failed delivery means the registry records an attempt and acks
+		// the message instead of nacking it into the DLQ.
+		require.NotNil(t, delivery)
+		assert.Equal(t, "failed", delivery.Status)
+		assert.Equal(t, "ERR", delivery.Code)
+
+		var pubErr *destregistry.ErrDestinationPublishAttempt
+		require.ErrorAs(t, err, &pubErr)
+		assert.Equal(t, "aws_s3", pubErr.Provider)
+		assert.Equal(t, "format_failed", pubErr.Data["error"])
+		// Raw Go error is carried on the error for logs/telemetry...
+		assert.Equal(t, rawErr, pubErr.Err)
+	})
+
+	t.Run("uses a generic message when none is given", func(t *testing.T) {
+		t.Parallel()
+
+		delivery, _ := destregistry.NewFormatError("aws_s3", "", rawErr)
+
+		// ...but is NOT persisted on the attempt; the customer-facing response is generic.
+		assert.Equal(t, "could not format event for delivery", delivery.Response["error"])
+		assert.NotContains(t, delivery.Response["error"], "key template")
+	})
+
+	t.Run("uses the provided message when given", func(t *testing.T) {
+		t.Parallel()
+
+		delivery, _ := destregistry.NewFormatError("aws_s3", "could not build S3 object key", rawErr)
+		assert.Equal(t, "could not build S3 object key", delivery.Response["error"])
+	})
+}

internal/destregistry/providers/destawskinesis/destawskinesis.go

@@ -258,14 +258,7 @@ func (p *AWSKinesisPublisher) Publish(ctx context.Context, event *models.Event)
 	// Format the event into a PutRecordInput
 	input, err := p.Format(ctx, event)
 	if err != nil {
-		return nil, destregistry.NewErrDestinationPublishAttempt(
-			err,
-			"aws_kinesis",
-			map[string]interface{}{
-				"error":   "format_failed",
-				"message": err.Error(),
-			},
-		)
+		return destregistry.NewFormatError("aws_kinesis", "", err)
 	}
 
 	// Send the record to Kinesis

internal/destregistry/providers/destawss3/destawss3.go

@@ -304,7 +304,7 @@ func (p *AWSS3Publisher) Publish(ctx context.Context, event *models.Event) (*des
 
 	input, err := p.Format(ctx, event)
 	if err != nil {
-		return nil, err
+		return destregistry.NewFormatError("aws_s3", "", err)
 	}
 
 	_, err = p.client.PutObject(ctx, input)

internal/destregistry/providers/destawss3/destawss3_format_test.go

@@ -198,6 +198,42 @@ func TestAWSS3Publisher_Format_NilResult(t *testing.T) {
 	assert.Contains(t, err.Error(), "nil result")
 }
 
+// TestAWSS3Publisher_Publish_FormatError reproduces the production incident where a
+// key_template references a field absent from the event (here metadata.operationId on
+// an event with no such metadata). The key cannot be built, so Format fails. Publish
+// must surface this as a failed delivery + ErrDestinationPublishAttempt (so the registry
+// records an attempt and acks) rather than a nil delivery (which nacks into the DLQ).
+func TestAWSS3Publisher_Publish_FormatError(t *testing.T) {
+	event := models.Event{
+		ID:   "event-123",
+		Time: time.Now(),
+		Data: json.RawMessage(`{}`),
+	}
+
+	// Mirrors the incident template: join over a missing metadata field yields nil,
+	// which join rejects. Reaches Publish's Format step before any S3 client call,
+	// so a nil client is safe here.
+	template := `join('/', ['prefix', metadata.operationId])`
+	publisher := destawss3.NewAWSS3Publisher(
+		destregistry.NewBasePublisher(),
+		nil,
+		"my-bucket",
+		template,
+		"STANDARD",
+	)
+
+	delivery, err := publisher.Publish(context.Background(), &event)
+
+	require.NotNil(t, delivery, "format failure must yield a non-nil delivery so the message is acked, not DLQ'd")
+	assert.Equal(t, "failed", delivery.Status)
+	assert.Equal(t, "ERR", delivery.Code)
+
+	var pubErr *destregistry.ErrDestinationPublishAttempt
+	require.ErrorAs(t, err, &pubErr)
+	assert.Equal(t, "aws_s3", pubErr.Provider)
+	assert.Equal(t, "format_failed", pubErr.Data["error"])
+}
+
 func TestAWSS3Publisher_Format_EmptyResult(t *testing.T) {
 	event := models.Event{
 		ID:   "event-123",

internal/destregistry/providers/destawssqs/destawssqs.go

@@ -163,7 +163,7 @@ func (p *AWSSQSPublisher) Publish(ctx context.Context, event *models.Event) (*de
 
 	msg, err := p.Format(ctx, event)
 	if err != nil {
-		return nil, err
+		return destregistry.NewFormatError("aws_sqs", "", err)
 	}
 
 	if _, err = p.client.SendMessage(ctx, msg); err != nil {

internal/destregistry/providers/destazureservicebus/destazureservicebus.go

@@ -149,7 +149,7 @@ func (p *AzureServiceBusPublisher) Publish(ctx context.Context, event *models.Ev
 
 	message, err := p.Format(ctx, event)
 	if err != nil {
-		return nil, err
+		return destregistry.NewFormatError("azure_servicebus", "", err)
 	}
 
 	sender, err := p.ensureSender()

internal/destregistry/providers/destgcppubsub/destgcppubsub.go

@@ -184,7 +184,7 @@ func (pub *GCPPubSubPublisher) Publish(ctx context.Context, event *models.Event)
 	// Format the message
 	msg, err := pub.Format(ctx, event)
 	if err != nil {
-		return nil, err
+		return destregistry.NewFormatError("gcp_pubsub", "", err)
 	}
 
 	// Publish the message

internal/destregistry/providers/destkafka/destkafka.go

@@ -222,12 +222,7 @@ func (p *KafkaPublisher) Publish(ctx context.Context, event *models.Event) (*des
 	// Build parsed payload for partition key JMESPath evaluation
 	dataMap, err := event.ParsedData()
 	if err != nil {
-		return nil, destregistry.NewErrDestinationPublishAttempt(
-			err, "kafka", map[string]interface{}{
-				"error":   "format_failed",
-				"message": err.Error(),
-			},
-		)
+		return destregistry.NewFormatError("kafka", "", err)
 	}
 	if dataMap == nil {
 		dataMap = make(map[string]interface{})

internal/destregistry/providers/destwebhook/destwebhook.go

@@ -658,7 +658,7 @@ func (p *WebhookPublisher) Publish(ctx context.Context, event *models.Event) (*d
 
 	httpReq, err := p.Format(ctx, event)
 	if err != nil {
-		return nil, err
+		return destregistry.NewFormatError("webhook", "", err)
 	}
 
 	result := ExecuteHTTPRequest(ctx, p.httpClient, httpReq, "webhook")

internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go

@@ -555,7 +555,7 @@ func (p *StandardWebhookPublisher) Publish(ctx context.Context, event *models.Ev
 
 	httpReq, err := p.Format(ctx, event)
 	if err != nil {
-		return nil, err
+		return destregistry.NewFormatError("webhook_standard", "", err)
 	}
 
 	result := destwebhook.ExecuteHTTPRequest(ctx, p.httpClient, httpReq, "webhook_standard")