diff --git a/cmd/e2e/regression_format_error_test.go b/cmd/e2e/regression_format_error_test.go new file mode 100644 index 000000000..eb2d8861e --- /dev/null +++ b/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") +} diff --git a/internal/destregistry/error.go b/internal/destregistry/error.go index c824d9b26..f07f576d1 100644 --- a/internal/destregistry/error.go +++ b/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 diff --git a/internal/destregistry/format_error_test.go b/internal/destregistry/format_error_test.go new file mode 100644 index 000000000..b2db7c095 --- /dev/null +++ b/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"]) + }) +} diff --git a/internal/destregistry/providers/destawskinesis/destawskinesis.go b/internal/destregistry/providers/destawskinesis/destawskinesis.go index 7d5ec3e5a..0baf29d18 100644 --- a/internal/destregistry/providers/destawskinesis/destawskinesis.go +++ b/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 diff --git a/internal/destregistry/providers/destawss3/destawss3.go b/internal/destregistry/providers/destawss3/destawss3.go index 1d7d80bce..6a30a1267 100644 --- a/internal/destregistry/providers/destawss3/destawss3.go +++ b/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) diff --git a/internal/destregistry/providers/destawss3/destawss3_format_test.go b/internal/destregistry/providers/destawss3/destawss3_format_test.go index e206c5c59..c197c784e 100644 --- a/internal/destregistry/providers/destawss3/destawss3_format_test.go +++ b/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", diff --git a/internal/destregistry/providers/destawssqs/destawssqs.go b/internal/destregistry/providers/destawssqs/destawssqs.go index f605e3b2c..511140b47 100644 --- a/internal/destregistry/providers/destawssqs/destawssqs.go +++ b/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 { diff --git a/internal/destregistry/providers/destazureservicebus/destazureservicebus.go b/internal/destregistry/providers/destazureservicebus/destazureservicebus.go index 65252bfd5..7034d991d 100644 --- a/internal/destregistry/providers/destazureservicebus/destazureservicebus.go +++ b/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() diff --git a/internal/destregistry/providers/destgcppubsub/destgcppubsub.go b/internal/destregistry/providers/destgcppubsub/destgcppubsub.go index f522411b4..9ee1e9adc 100644 --- a/internal/destregistry/providers/destgcppubsub/destgcppubsub.go +++ b/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 diff --git a/internal/destregistry/providers/destkafka/destkafka.go b/internal/destregistry/providers/destkafka/destkafka.go index a75cdc637..6d7cfafaa 100644 --- a/internal/destregistry/providers/destkafka/destkafka.go +++ b/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{}) diff --git a/internal/destregistry/providers/destwebhook/destwebhook.go b/internal/destregistry/providers/destwebhook/destwebhook.go index 0162a2e23..055a4318e 100644 --- a/internal/destregistry/providers/destwebhook/destwebhook.go +++ b/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") diff --git a/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go b/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go index fc4f098b1..33d47c740 100644 --- a/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go +++ b/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")